mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-19 02:56:18 +00:00
refactor: narrow IAM and Swift compatibility surfaces (#3587)
* refactor: narrow IAM and Swift compatibility surfaces * refactor: narrow heal and scanner compatibility surfaces (#3588) * refactor: narrow RustFS runtime compatibility surfaces (#3591)
This commit is contained in:
@@ -12,19 +12,22 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
pub(crate) mod ecstore {
|
||||
pub(crate) use rustfs_ecstore::{data_usage, disk, error, global, store};
|
||||
}
|
||||
pub(crate) use self::ecstore::{
|
||||
data_usage::DATA_USAGE_CACHE_NAME,
|
||||
disk::{BUCKET_META_PREFIX, DiskAPI, DiskStore, RUSTFS_META_BUCKET, endpoint::Endpoint, error::DiskError},
|
||||
error::{Error as EcstoreError, StorageError},
|
||||
global::GLOBAL_LOCAL_DISK_MAP,
|
||||
store::ECStore,
|
||||
};
|
||||
pub(crate) const DATA_USAGE_CACHE_NAME: &str = rustfs_ecstore::data_usage::DATA_USAGE_CACHE_NAME;
|
||||
pub(crate) const BUCKET_META_PREFIX: &str = rustfs_ecstore::disk::BUCKET_META_PREFIX;
|
||||
pub(crate) const RUSTFS_META_BUCKET: &str = rustfs_ecstore::disk::RUSTFS_META_BUCKET;
|
||||
|
||||
pub(crate) type DiskError = rustfs_ecstore::disk::error::DiskError;
|
||||
pub(crate) type DiskStore = rustfs_ecstore::disk::DiskStore;
|
||||
pub(crate) type ECStore = rustfs_ecstore::store::ECStore;
|
||||
pub(crate) type EcstoreError = rustfs_ecstore::error::Error;
|
||||
pub(crate) type Endpoint = rustfs_ecstore::disk::endpoint::Endpoint;
|
||||
pub(crate) type StorageError = rustfs_ecstore::error::StorageError;
|
||||
|
||||
pub(crate) use rustfs_ecstore::disk::DiskAPI;
|
||||
pub(crate) use rustfs_ecstore::global::GLOBAL_LOCAL_DISK_MAP;
|
||||
|
||||
#[cfg(test)]
|
||||
pub(crate) use self::ecstore::disk::{DiskOption, new_disk};
|
||||
pub(crate) use rustfs_ecstore::disk::{DiskOption, new_disk};
|
||||
|
||||
pub type HealObjectInfo = rustfs_ecstore::store_api::ObjectInfo;
|
||||
pub type HealObjectOptions = rustfs_ecstore::store_api::ObjectOptions;
|
||||
|
||||
+10
-9
@@ -12,6 +12,7 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use crate::storage_compat::IamStorageError;
|
||||
use rustfs_policy::policy::Error as PolicyError;
|
||||
|
||||
pub type Result<T> = core::result::Result<T, Error>;
|
||||
@@ -179,20 +180,20 @@ impl Error {
|
||||
}
|
||||
}
|
||||
|
||||
impl From<crate::storage_compat::ecstore::error::StorageError> for Error {
|
||||
fn from(e: crate::storage_compat::ecstore::error::StorageError) -> Self {
|
||||
impl From<IamStorageError> for Error {
|
||||
fn from(e: IamStorageError) -> Self {
|
||||
match e {
|
||||
crate::storage_compat::ecstore::error::StorageError::ConfigNotFound => Error::ConfigNotFound,
|
||||
IamStorageError::ConfigNotFound => Error::ConfigNotFound,
|
||||
_ => Error::other(e),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl From<Error> for crate::storage_compat::ecstore::error::StorageError {
|
||||
impl From<Error> for IamStorageError {
|
||||
fn from(e: Error) -> Self {
|
||||
match e {
|
||||
Error::ConfigNotFound => crate::storage_compat::ecstore::error::StorageError::ConfigNotFound,
|
||||
_ => crate::storage_compat::ecstore::error::StorageError::other(e),
|
||||
Error::ConfigNotFound => IamStorageError::ConfigNotFound,
|
||||
_ => IamStorageError::other(e),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -330,13 +331,13 @@ mod tests {
|
||||
#[test]
|
||||
fn test_iam_error_from_storage_error() {
|
||||
// Test conversion from StorageError
|
||||
let storage_error = crate::storage_compat::ecstore::error::StorageError::ConfigNotFound;
|
||||
let storage_error = IamStorageError::ConfigNotFound;
|
||||
let iam_error: Error = storage_error.into();
|
||||
assert_eq!(iam_error, Error::ConfigNotFound);
|
||||
|
||||
// Test reverse conversion
|
||||
let back_to_storage: crate::storage_compat::ecstore::error::StorageError = iam_error.into();
|
||||
assert_eq!(back_to_storage, crate::storage_compat::ecstore::error::StorageError::ConfigNotFound);
|
||||
let back_to_storage: IamStorageError = iam_error.into();
|
||||
assert_eq!(back_to_storage, IamStorageError::ConfigNotFound);
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
@@ -13,7 +13,7 @@
|
||||
// limitations under the License.
|
||||
|
||||
use crate::error::{Error, Result};
|
||||
use crate::storage_compat::ecstore::store::ECStore;
|
||||
use crate::storage_compat::IamStore;
|
||||
use manager::IamCache;
|
||||
use oidc::OidcSys;
|
||||
use std::sync::{Arc, OnceLock};
|
||||
@@ -42,7 +42,7 @@ static IAM_SYS: OnceLock<Arc<IamSys<ObjectStore>>> = OnceLock::new();
|
||||
static OIDC_SYS: OnceLock<Arc<OidcSys>> = OnceLock::new();
|
||||
|
||||
#[instrument(skip(ecstore))]
|
||||
pub async fn init_iam_sys(ecstore: Arc<ECStore>) -> Result<()> {
|
||||
pub async fn init_iam_sys(ecstore: Arc<IamStore>) -> Result<()> {
|
||||
if IAM_SYS.get().is_some() {
|
||||
info!(
|
||||
event = EVENT_IAM_STATE,
|
||||
|
||||
@@ -13,7 +13,7 @@
|
||||
// limitations under the License.
|
||||
|
||||
use crate::error::{Error, Result, is_err_config_not_found};
|
||||
use crate::storage_compat::ecstore::global::is_first_cluster_node_local;
|
||||
use crate::storage_compat::is_iam_first_cluster_node_local;
|
||||
use crate::sys::{get_claims_from_token_with_secret, get_claims_from_token_with_secret_allow_missing_exp};
|
||||
use crate::{
|
||||
cache::{Cache, CacheEntity, LockedCache},
|
||||
@@ -375,7 +375,7 @@ where
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
if !is_first_cluster_node_local().await {
|
||||
if !is_iam_first_cluster_node_local().await {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
|
||||
@@ -12,6 +12,155 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
pub(crate) mod ecstore {
|
||||
pub(crate) use rustfs_ecstore::{config, error, global, notification_sys, store};
|
||||
use std::sync::Arc;
|
||||
|
||||
pub(crate) const IAM_CONFIG_ROOT_PREFIX: &str = rustfs_ecstore::config::RUSTFS_CONFIG_PREFIX;
|
||||
|
||||
pub(crate) type IamEcstoreError = rustfs_ecstore::error::Error;
|
||||
pub(crate) type IamConfigObjectInfo = rustfs_ecstore::store_api::ObjectInfo;
|
||||
pub(crate) type IamConfigObjectOptions = rustfs_ecstore::store_api::ObjectOptions;
|
||||
pub(crate) type IamStorageError = rustfs_ecstore::error::StorageError;
|
||||
pub(crate) type IamStorageResult<T> = rustfs_ecstore::error::Result<T>;
|
||||
pub(crate) type IamStore = rustfs_ecstore::store::ECStore;
|
||||
|
||||
pub(crate) async fn read_iam_config_no_lock(api: Arc<IamStore>, file: &str) -> IamStorageResult<Vec<u8>> {
|
||||
rustfs_ecstore::config::com::read_config_no_lock(api, file).await
|
||||
}
|
||||
|
||||
pub(crate) async fn read_iam_config_with_metadata(
|
||||
api: Arc<IamStore>,
|
||||
file: &str,
|
||||
opts: &IamConfigObjectOptions,
|
||||
) -> IamStorageResult<(Vec<u8>, IamConfigObjectInfo)> {
|
||||
rustfs_ecstore::config::com::read_config_with_metadata(api, file, opts).await
|
||||
}
|
||||
|
||||
pub(crate) async fn save_iam_config(api: Arc<IamStore>, file: &str, data: Vec<u8>) -> IamStorageResult<()> {
|
||||
rustfs_ecstore::config::com::save_config(api, file, data).await
|
||||
}
|
||||
|
||||
pub(crate) async fn save_iam_config_with_opts(
|
||||
api: Arc<IamStore>,
|
||||
file: &str,
|
||||
data: Vec<u8>,
|
||||
opts: &IamConfigObjectOptions,
|
||||
) -> IamStorageResult<()> {
|
||||
rustfs_ecstore::config::com::save_config_with_opts(api, file, data, opts).await
|
||||
}
|
||||
|
||||
pub(crate) async fn delete_iam_config(api: Arc<IamStore>, file: &str) -> IamStorageResult<()> {
|
||||
rustfs_ecstore::config::com::delete_config(api, file).await
|
||||
}
|
||||
|
||||
pub(crate) fn classify_iam_system_path_failure_reason(err: &IamEcstoreError) -> &'static str {
|
||||
rustfs_ecstore::error::classify_system_path_failure_reason(err)
|
||||
}
|
||||
|
||||
pub(crate) async fn is_iam_first_cluster_node_local() -> bool {
|
||||
rustfs_ecstore::global::is_first_cluster_node_local().await
|
||||
}
|
||||
|
||||
pub(crate) struct IamNotificationPeerErr {
|
||||
pub(crate) err: Option<IamEcstoreError>,
|
||||
}
|
||||
|
||||
impl From<rustfs_ecstore::notification_sys::NotificationPeerErr> for IamNotificationPeerErr {
|
||||
fn from(value: rustfs_ecstore::notification_sys::NotificationPeerErr) -> Self {
|
||||
Self { err: value.err }
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) async fn notify_iam_delete_policy(policy_name: &str) -> Vec<IamNotificationPeerErr> {
|
||||
match rustfs_ecstore::notification_sys::get_global_notification_sys() {
|
||||
Some(notification_sys) => notification_sys
|
||||
.delete_policy(policy_name)
|
||||
.await
|
||||
.into_iter()
|
||||
.map(Into::into)
|
||||
.collect(),
|
||||
None => Vec::new(),
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) async fn notify_iam_load_policy(policy_name: &str) -> Vec<IamNotificationPeerErr> {
|
||||
match rustfs_ecstore::notification_sys::get_global_notification_sys() {
|
||||
Some(notification_sys) => notification_sys
|
||||
.load_policy(policy_name)
|
||||
.await
|
||||
.into_iter()
|
||||
.map(Into::into)
|
||||
.collect(),
|
||||
None => Vec::new(),
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) async fn notify_iam_delete_user(access_key: &str) -> Vec<IamNotificationPeerErr> {
|
||||
match rustfs_ecstore::notification_sys::get_global_notification_sys() {
|
||||
Some(notification_sys) => notification_sys
|
||||
.delete_user(access_key)
|
||||
.await
|
||||
.into_iter()
|
||||
.map(Into::into)
|
||||
.collect(),
|
||||
None => Vec::new(),
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) async fn notify_iam_load_user(access_key: &str, temp: bool) -> Vec<IamNotificationPeerErr> {
|
||||
match rustfs_ecstore::notification_sys::get_global_notification_sys() {
|
||||
Some(notification_sys) => notification_sys
|
||||
.load_user(access_key, temp)
|
||||
.await
|
||||
.into_iter()
|
||||
.map(Into::into)
|
||||
.collect(),
|
||||
None => Vec::new(),
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) async fn notify_iam_load_service_account(access_key: &str) -> Vec<IamNotificationPeerErr> {
|
||||
match rustfs_ecstore::notification_sys::get_global_notification_sys() {
|
||||
Some(notification_sys) => notification_sys
|
||||
.load_service_account(access_key)
|
||||
.await
|
||||
.into_iter()
|
||||
.map(Into::into)
|
||||
.collect(),
|
||||
None => Vec::new(),
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) async fn notify_iam_delete_service_account(access_key: &str) -> Vec<IamNotificationPeerErr> {
|
||||
match rustfs_ecstore::notification_sys::get_global_notification_sys() {
|
||||
Some(notification_sys) => notification_sys
|
||||
.delete_service_account(access_key)
|
||||
.await
|
||||
.into_iter()
|
||||
.map(Into::into)
|
||||
.collect(),
|
||||
None => Vec::new(),
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) async fn notify_iam_load_group(group: &str) -> Vec<IamNotificationPeerErr> {
|
||||
match rustfs_ecstore::notification_sys::get_global_notification_sys() {
|
||||
Some(notification_sys) => notification_sys.load_group(group).await.into_iter().map(Into::into).collect(),
|
||||
None => Vec::new(),
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) async fn notify_iam_load_policy_mapping(
|
||||
user_or_group: &str,
|
||||
user_type: u64,
|
||||
is_group: bool,
|
||||
) -> Vec<IamNotificationPeerErr> {
|
||||
match rustfs_ecstore::notification_sys::get_global_notification_sys() {
|
||||
Some(notification_sys) => notification_sys
|
||||
.load_policy_mapping(user_or_group, user_type, is_group)
|
||||
.await
|
||||
.into_iter()
|
||||
.map(Into::into)
|
||||
.collect(),
|
||||
None => Vec::new(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -14,13 +14,9 @@
|
||||
|
||||
use super::{GroupInfo, MappedPolicy, Store, UserType};
|
||||
use crate::error::{Error, Result, is_err_config_not_found, is_err_no_such_group};
|
||||
use crate::storage_compat::ecstore::error::{Error as EcstoreError, StorageError, classify_system_path_failure_reason};
|
||||
use crate::storage_compat::ecstore::{
|
||||
config::{
|
||||
RUSTFS_CONFIG_PREFIX,
|
||||
com::{delete_config, read_config_no_lock, read_config_with_metadata, save_config, save_config_with_opts},
|
||||
},
|
||||
store::ECStore,
|
||||
use crate::storage_compat::{
|
||||
IAM_CONFIG_ROOT_PREFIX, IamStorageError, IamStore, classify_iam_system_path_failure_reason, delete_iam_config,
|
||||
read_iam_config_no_lock, read_iam_config_with_metadata, save_iam_config, save_iam_config_with_opts,
|
||||
};
|
||||
use crate::{
|
||||
cache::{Cache, CacheEntity},
|
||||
@@ -44,24 +40,24 @@ use tracing::{debug, error, warn};
|
||||
|
||||
use super::storage_compat::{IamObjectInfo, IamObjectOptions};
|
||||
|
||||
pub static IAM_CONFIG_PREFIX: LazyLock<String> = LazyLock::new(|| format!("{RUSTFS_CONFIG_PREFIX}/iam"));
|
||||
pub static IAM_CONFIG_USERS_PREFIX: LazyLock<String> = LazyLock::new(|| format!("{RUSTFS_CONFIG_PREFIX}/iam/users/"));
|
||||
pub static IAM_CONFIG_PREFIX: LazyLock<String> = LazyLock::new(|| format!("{IAM_CONFIG_ROOT_PREFIX}/iam"));
|
||||
pub static IAM_CONFIG_USERS_PREFIX: LazyLock<String> = LazyLock::new(|| format!("{IAM_CONFIG_ROOT_PREFIX}/iam/users/"));
|
||||
pub static IAM_CONFIG_SERVICE_ACCOUNTS_PREFIX: LazyLock<String> =
|
||||
LazyLock::new(|| format!("{RUSTFS_CONFIG_PREFIX}/iam/service-accounts/"));
|
||||
pub static IAM_CONFIG_GROUPS_PREFIX: LazyLock<String> = LazyLock::new(|| format!("{RUSTFS_CONFIG_PREFIX}/iam/groups/"));
|
||||
pub static IAM_CONFIG_POLICIES_PREFIX: LazyLock<String> = LazyLock::new(|| format!("{RUSTFS_CONFIG_PREFIX}/iam/policies/"));
|
||||
pub static IAM_CONFIG_STS_PREFIX: LazyLock<String> = LazyLock::new(|| format!("{RUSTFS_CONFIG_PREFIX}/iam/sts/"));
|
||||
pub static IAM_CONFIG_POLICY_DB_PREFIX: LazyLock<String> = LazyLock::new(|| format!("{RUSTFS_CONFIG_PREFIX}/iam/policydb/"));
|
||||
LazyLock::new(|| format!("{IAM_CONFIG_ROOT_PREFIX}/iam/service-accounts/"));
|
||||
pub static IAM_CONFIG_GROUPS_PREFIX: LazyLock<String> = LazyLock::new(|| format!("{IAM_CONFIG_ROOT_PREFIX}/iam/groups/"));
|
||||
pub static IAM_CONFIG_POLICIES_PREFIX: LazyLock<String> = LazyLock::new(|| format!("{IAM_CONFIG_ROOT_PREFIX}/iam/policies/"));
|
||||
pub static IAM_CONFIG_STS_PREFIX: LazyLock<String> = LazyLock::new(|| format!("{IAM_CONFIG_ROOT_PREFIX}/iam/sts/"));
|
||||
pub static IAM_CONFIG_POLICY_DB_PREFIX: LazyLock<String> = LazyLock::new(|| format!("{IAM_CONFIG_ROOT_PREFIX}/iam/policydb/"));
|
||||
pub static IAM_CONFIG_POLICY_DB_USERS_PREFIX: LazyLock<String> =
|
||||
LazyLock::new(|| format!("{RUSTFS_CONFIG_PREFIX}/iam/policydb/users/"));
|
||||
LazyLock::new(|| format!("{IAM_CONFIG_ROOT_PREFIX}/iam/policydb/users/"));
|
||||
pub static IAM_CONFIG_POLICY_DB_STS_USERS_PREFIX: LazyLock<String> =
|
||||
LazyLock::new(|| format!("{RUSTFS_CONFIG_PREFIX}/iam/policydb/sts-users/"));
|
||||
LazyLock::new(|| format!("{IAM_CONFIG_ROOT_PREFIX}/iam/policydb/sts-users/"));
|
||||
pub static IAM_CONFIG_POLICY_DB_SERVICE_ACCOUNTS_PREFIX: LazyLock<String> =
|
||||
LazyLock::new(|| format!("{RUSTFS_CONFIG_PREFIX}/iam/policydb/service-accounts/"));
|
||||
LazyLock::new(|| format!("{IAM_CONFIG_ROOT_PREFIX}/iam/policydb/service-accounts/"));
|
||||
pub static IAM_CONFIG_POLICY_DB_GROUPS_PREFIX: LazyLock<String> =
|
||||
LazyLock::new(|| format!("{RUSTFS_CONFIG_PREFIX}/iam/policydb/groups/"));
|
||||
LazyLock::new(|| format!("{IAM_CONFIG_ROOT_PREFIX}/iam/policydb/groups/"));
|
||||
|
||||
type ObjectInfoOrErr = StorageObjectInfoOrErr<IamObjectInfo, EcstoreError>;
|
||||
type ObjectInfoOrErr = StorageObjectInfoOrErr<IamObjectInfo, IamStorageError>;
|
||||
|
||||
const IAM_IDENTITY_FILE: &str = "identity.json";
|
||||
const IAM_POLICY_FILE: &str = "policy.json";
|
||||
@@ -150,13 +146,13 @@ fn split_path(s: &str, last_index: bool) -> (&str, &str) {
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct ObjectStore {
|
||||
object_api: Arc<ECStore>,
|
||||
object_api: Arc<IamStore>,
|
||||
}
|
||||
|
||||
impl ObjectStore {
|
||||
const BUCKET_NAME: &'static str = ".rustfs.sys";
|
||||
|
||||
pub fn new(object_api: Arc<ECStore>) -> Self {
|
||||
pub fn new(object_api: Arc<IamStore>) -> Self {
|
||||
Self { object_api }
|
||||
}
|
||||
|
||||
@@ -308,7 +304,7 @@ impl ObjectStore {
|
||||
Ok(_) => {
|
||||
Self::complete_lazy_rewrite(path.as_str(), true);
|
||||
}
|
||||
Err(StorageError::PreconditionFailed) => {
|
||||
Err(IamStorageError::PreconditionFailed) => {
|
||||
Self::complete_lazy_rewrite(path.as_str(), false);
|
||||
debug!(path = %path, state = "stale_etag", "IAM lazy rewrite skipped");
|
||||
}
|
||||
@@ -320,8 +316,8 @@ impl ObjectStore {
|
||||
});
|
||||
}
|
||||
|
||||
async fn lazy_rewrite_iam_config(&self, path: &str, plain: &[u8], etag: &str) -> std::result::Result<(), StorageError> {
|
||||
let encrypted = Self::encrypt_data_with_master_key(plain).map_err(StorageError::other)?;
|
||||
async fn lazy_rewrite_iam_config(&self, path: &str, plain: &[u8], etag: &str) -> std::result::Result<(), IamStorageError> {
|
||||
let encrypted = Self::encrypt_data_with_master_key(plain).map_err(IamStorageError::other)?;
|
||||
|
||||
let mut opts = IamObjectOptions {
|
||||
max_parity: true,
|
||||
@@ -332,7 +328,7 @@ impl ObjectStore {
|
||||
..Default::default()
|
||||
});
|
||||
|
||||
save_config_with_opts(self.object_api.clone(), path, encrypted, &opts).await
|
||||
save_iam_config_with_opts(self.object_api.clone(), path, encrypted, &opts).await
|
||||
}
|
||||
|
||||
fn is_plaintext_json(data: &[u8]) -> bool {
|
||||
@@ -346,7 +342,7 @@ impl ObjectStore {
|
||||
|
||||
async fn load_iamconfig_bytes_with_metadata(&self, path: impl AsRef<str> + Send) -> Result<(Vec<u8>, IamObjectInfo)> {
|
||||
let path_ref = path.as_ref();
|
||||
let (data, obj) = read_config_with_metadata(self.object_api.clone(), path_ref, &IamObjectOptions::default()).await?;
|
||||
let (data, obj) = read_iam_config_with_metadata(self.object_api.clone(), path_ref, &IamObjectOptions::default()).await?;
|
||||
let outcome = Self::decrypt_data_with_source(&data)?;
|
||||
self.maybe_schedule_lazy_rewrite(path_ref, &outcome, &obj);
|
||||
|
||||
@@ -371,7 +367,7 @@ impl ObjectStore {
|
||||
.walk(ctx.clone(), Self::BUCKET_NAME, &path, tx, Default::default())
|
||||
.await
|
||||
{
|
||||
let reason = classify_system_path_failure_reason(&err);
|
||||
let reason = classify_iam_system_path_failure_reason(&err);
|
||||
record_system_path_failure("iam_config", "walk", reason);
|
||||
error!(
|
||||
path_kind = "iam_config",
|
||||
@@ -587,9 +583,9 @@ impl ObjectStore {
|
||||
// If it doesn't exist, the system bucket or metadata is not ready.
|
||||
let probe_path = format!("{}/format.json", *IAM_CONFIG_PREFIX);
|
||||
|
||||
match read_config_no_lock(self.object_api.clone(), &probe_path).await {
|
||||
match read_iam_config_no_lock(self.object_api.clone(), &probe_path).await {
|
||||
Ok(_) => Ok(()),
|
||||
Err(crate::storage_compat::ecstore::error::StorageError::ConfigNotFound) => Err(Error::other(format!(
|
||||
Err(IamStorageError::ConfigNotFound) => Err(Error::other(format!(
|
||||
"Storage metadata not ready: probe object '{}' not found (expected IAM config to be initialized)",
|
||||
probe_path
|
||||
))),
|
||||
@@ -641,7 +637,7 @@ impl Store for ObjectStore {
|
||||
}
|
||||
async fn load_iam_config<Item: DeserializeOwned>(&self, path: impl AsRef<str> + Send) -> Result<Item> {
|
||||
let path_ref = path.as_ref();
|
||||
let (data, obj) = read_config_with_metadata(self.object_api.clone(), path_ref, &IamObjectOptions::default()).await?;
|
||||
let (data, obj) = read_iam_config_with_metadata(self.object_api.clone(), path_ref, &IamObjectOptions::default()).await?;
|
||||
|
||||
let outcome = match Self::decrypt_data_with_source(&data) {
|
||||
Ok(v) => v,
|
||||
@@ -679,7 +675,7 @@ impl Store for ObjectStore {
|
||||
let path_ref = path.as_ref();
|
||||
|
||||
loop {
|
||||
match save_config(self.object_api.clone(), path_ref, data.clone()).await {
|
||||
match save_iam_config(self.object_api.clone(), path_ref, data.clone()).await {
|
||||
Ok(_) => {
|
||||
debug!("Successfully saved IAM config to {}", path_ref);
|
||||
return Ok(());
|
||||
@@ -702,7 +698,7 @@ impl Store for ObjectStore {
|
||||
}
|
||||
}
|
||||
async fn delete_iam_config(&self, path: impl AsRef<str> + Send) -> Result<()> {
|
||||
delete_config(self.object_api.clone(), path.as_ref()).await?;
|
||||
delete_iam_config(self.object_api.clone(), path.as_ref()).await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
||||
+24
-47
@@ -19,7 +19,10 @@ use crate::error::{Error, Result};
|
||||
use crate::manager::extract_jwt_claims;
|
||||
use crate::manager::get_default_policyes;
|
||||
use crate::manager::{IamCache, IamSyncMetricsSnapshot};
|
||||
use crate::storage_compat::ecstore::notification_sys::get_global_notification_sys;
|
||||
use crate::storage_compat::{
|
||||
notify_iam_delete_policy, notify_iam_delete_service_account, notify_iam_delete_user, notify_iam_load_group,
|
||||
notify_iam_load_policy, notify_iam_load_policy_mapping, notify_iam_load_service_account, notify_iam_load_user,
|
||||
};
|
||||
use crate::store::GroupInfo;
|
||||
use crate::store::MappedPolicy;
|
||||
use crate::store::Store;
|
||||
@@ -249,12 +252,9 @@ impl<T: Store> IamSys<T> {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
if let Some(notification_sys) = get_global_notification_sys() {
|
||||
let resp = notification_sys.delete_policy(name).await;
|
||||
for r in resp {
|
||||
if let Some(err) = r.err {
|
||||
warn!("notify delete_policy failed: {}", err);
|
||||
}
|
||||
for r in notify_iam_delete_policy(name).await {
|
||||
if let Some(err) = r.err {
|
||||
warn!("notify delete_policy failed: {}", err);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -293,11 +293,8 @@ impl<T: Store> IamSys<T> {
|
||||
pub async fn set_policy(&self, name: &str, policy: Policy) -> Result<OffsetDateTime> {
|
||||
let updated_at = self.store.set_policy(name, policy).await?;
|
||||
|
||||
if !self.has_watcher()
|
||||
&& let Some(notification_sys) = get_global_notification_sys()
|
||||
{
|
||||
let resp = notification_sys.load_policy(name).await;
|
||||
for r in resp {
|
||||
if !self.has_watcher() {
|
||||
for r in notify_iam_load_policy(name).await {
|
||||
if let Some(err) = r.err {
|
||||
warn!("notify load_policy failed: {}", err);
|
||||
}
|
||||
@@ -322,12 +319,8 @@ impl<T: Store> IamSys<T> {
|
||||
pub async fn delete_user(&self, name: &str, notify: bool) -> Result<()> {
|
||||
self.store.delete_user(name, UserType::Reg).await?;
|
||||
|
||||
if notify
|
||||
&& !self.has_watcher()
|
||||
&& let Some(notification_sys) = get_global_notification_sys()
|
||||
{
|
||||
let resp = notification_sys.delete_user(name).await;
|
||||
for r in resp {
|
||||
if notify && !self.has_watcher() {
|
||||
for r in notify_iam_delete_user(name).await {
|
||||
if let Some(err) = r.err {
|
||||
warn!("notify delete_user failed: {}", err);
|
||||
}
|
||||
@@ -346,12 +339,9 @@ impl<T: Store> IamSys<T> {
|
||||
// This is critical for cluster recovery: login should not wait for dead peers
|
||||
let name = name.to_string();
|
||||
tokio::spawn(async move {
|
||||
if let Some(notification_sys) = get_global_notification_sys() {
|
||||
let resp = notification_sys.load_user(&name, is_temp).await;
|
||||
for r in resp {
|
||||
if let Some(err) = r.err {
|
||||
warn!("notify load_user failed (non-blocking): {}", err);
|
||||
}
|
||||
for r in notify_iam_load_user(&name, is_temp).await {
|
||||
if let Some(err) = r.err {
|
||||
warn!("notify load_user failed (non-blocking): {}", err);
|
||||
}
|
||||
}
|
||||
});
|
||||
@@ -365,12 +355,9 @@ impl<T: Store> IamSys<T> {
|
||||
// Fire-and-forget notification to peers - don't block service account operations
|
||||
let name = name.to_string();
|
||||
tokio::spawn(async move {
|
||||
if let Some(notification_sys) = get_global_notification_sys() {
|
||||
let resp = notification_sys.load_service_account(&name).await;
|
||||
for r in resp {
|
||||
if let Some(err) = r.err {
|
||||
warn!("notify load_service_account failed (non-blocking): {}", err);
|
||||
}
|
||||
for r in notify_iam_load_service_account(&name).await {
|
||||
if let Some(err) = r.err {
|
||||
warn!("notify load_service_account failed (non-blocking): {}", err);
|
||||
}
|
||||
}
|
||||
});
|
||||
@@ -687,12 +674,8 @@ impl<T: Store> IamSys<T> {
|
||||
|
||||
self.store.delete_user(access_key, UserType::Svc).await?;
|
||||
|
||||
if notify
|
||||
&& !self.has_watcher()
|
||||
&& let Some(notification_sys) = get_global_notification_sys()
|
||||
{
|
||||
let resp = notification_sys.delete_service_account(access_key).await;
|
||||
for r in resp {
|
||||
if notify && !self.has_watcher() {
|
||||
for r in notify_iam_delete_service_account(access_key).await {
|
||||
if let Some(err) = r.err {
|
||||
warn!("notify delete_service_account failed: {}", err);
|
||||
}
|
||||
@@ -710,12 +693,9 @@ impl<T: Store> IamSys<T> {
|
||||
// Fire-and-forget notification to peers - don't block group operations
|
||||
let group = group.to_string();
|
||||
tokio::spawn(async move {
|
||||
if let Some(notification_sys) = get_global_notification_sys() {
|
||||
let resp = notification_sys.load_group(&group).await;
|
||||
for r in resp {
|
||||
if let Some(err) = r.err {
|
||||
warn!("notify load_group failed (non-blocking): {}", err);
|
||||
}
|
||||
for r in notify_iam_load_group(&group).await {
|
||||
if let Some(err) = r.err {
|
||||
warn!("notify load_group failed (non-blocking): {}", err);
|
||||
}
|
||||
}
|
||||
});
|
||||
@@ -842,11 +822,8 @@ impl<T: Store> IamSys<T> {
|
||||
pub async fn policy_db_set(&self, name: &str, user_type: UserType, is_group: bool, policy: &str) -> Result<OffsetDateTime> {
|
||||
let updated_at = self.store.policy_db_set(name, user_type, is_group, policy).await?;
|
||||
|
||||
if !self.has_watcher()
|
||||
&& let Some(notification_sys) = get_global_notification_sys()
|
||||
{
|
||||
let resp = notification_sys.load_policy_mapping(name, user_type.to_u64(), is_group).await;
|
||||
for r in resp {
|
||||
if !self.has_watcher() {
|
||||
for r in notify_iam_load_policy_mapping(name, user_type.to_u64(), is_group).await {
|
||||
if let Some(err) = r.err {
|
||||
warn!("notify load_policy failed: {}", err);
|
||||
}
|
||||
|
||||
@@ -12,29 +12,25 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
mod ecstore {
|
||||
pub(super) use rustfs_ecstore::{bucket, error, resolve_object_store_handle, store};
|
||||
}
|
||||
use self::ecstore::{
|
||||
bucket::{metadata::BucketMetadata, metadata_sys},
|
||||
error::Result as EcstoreResult,
|
||||
store::ECStore,
|
||||
};
|
||||
use std::sync::Arc;
|
||||
|
||||
pub(super) type SwiftBucketMetadata = rustfs_ecstore::bucket::metadata::BucketMetadata;
|
||||
pub(super) type SwiftStorageResult<T> = rustfs_ecstore::error::Result<T>;
|
||||
pub(super) type SwiftStore = rustfs_ecstore::store::ECStore;
|
||||
|
||||
pub type SwiftGetObjectReader = rustfs_ecstore::store_api::GetObjectReader;
|
||||
pub type SwiftObjectInfo = rustfs_ecstore::store_api::ObjectInfo;
|
||||
pub type SwiftObjectOptions = rustfs_ecstore::store_api::ObjectOptions;
|
||||
pub type SwiftPutObjReader = rustfs_ecstore::store_api::PutObjReader;
|
||||
|
||||
pub fn resolve_swift_object_store_handle() -> Option<Arc<ECStore>> {
|
||||
ecstore::resolve_object_store_handle()
|
||||
pub fn resolve_swift_object_store_handle() -> Option<Arc<SwiftStore>> {
|
||||
rustfs_ecstore::resolve_object_store_handle()
|
||||
}
|
||||
|
||||
pub async fn get_swift_bucket_metadata(bucket: &str) -> EcstoreResult<Arc<BucketMetadata>> {
|
||||
metadata_sys::get(bucket).await
|
||||
pub async fn get_swift_bucket_metadata(bucket: &str) -> SwiftStorageResult<Arc<SwiftBucketMetadata>> {
|
||||
rustfs_ecstore::bucket::metadata_sys::get(bucket).await
|
||||
}
|
||||
|
||||
pub async fn set_swift_bucket_metadata(bucket: String, metadata: BucketMetadata) -> EcstoreResult<()> {
|
||||
metadata_sys::set_bucket_metadata(bucket, metadata).await
|
||||
pub async fn set_swift_bucket_metadata(bucket: String, metadata: SwiftBucketMetadata) -> SwiftStorageResult<()> {
|
||||
rustfs_ecstore::bucket::metadata_sys::set_bucket_metadata(bucket, metadata).await
|
||||
}
|
||||
|
||||
@@ -13,50 +13,50 @@
|
||||
// limitations under the License.
|
||||
|
||||
use http::HeaderMap;
|
||||
pub(crate) mod ecstore {
|
||||
pub(crate) use rustfs_ecstore::{
|
||||
bucket, cache_value, config, data_usage, disk, error, global, pools, resolve_object_store_handle, set_disk, store,
|
||||
store_utils,
|
||||
};
|
||||
}
|
||||
pub(crate) use self::ecstore::{
|
||||
bucket::{
|
||||
bucket_target_sys::BucketTargetSys,
|
||||
lifecycle::{
|
||||
bucket_lifecycle_audit::LcEventSrc,
|
||||
bucket_lifecycle_ops::{GLOBAL_ExpiryState, apply_expiry_rule, apply_transition_rule},
|
||||
evaluator::Evaluator,
|
||||
lifecycle::{Event, Lifecycle, ObjectOpts, TRANSITION_COMPLETE},
|
||||
},
|
||||
metadata_sys::{get_lifecycle_config, get_object_lock_config, get_replication_config},
|
||||
replication::{
|
||||
ReplicationConfig, ReplicationConfigurationExt, ReplicationQueueAdmission, queue_replication_heal_internal,
|
||||
},
|
||||
versioning::VersioningApi,
|
||||
versioning_sys::BucketVersioningSys,
|
||||
},
|
||||
cache_value::metacache_set::{ListPathRawOptions, list_path_raw},
|
||||
config::{
|
||||
com::{read_config, save_config},
|
||||
storageclass,
|
||||
},
|
||||
data_usage::replace_bucket_usage_memory_from_info,
|
||||
disk::{BUCKET_META_PREFIX, Disk, DiskAPI, DiskInfoOptions, RUSTFS_META_BUCKET, STORAGE_FORMAT_FILE, error::DiskError},
|
||||
error::{Error as EcstoreError, Result as EcstoreResult, StorageError},
|
||||
global::{GLOBAL_TierConfigMgr, is_erasure, is_erasure_sd},
|
||||
pools::{path2_bucket_object, path2_bucket_object_with_base_path},
|
||||
set_disk::SetDisks,
|
||||
store::ECStore,
|
||||
store_utils::is_reserved_or_invalid_bucket,
|
||||
};
|
||||
use rustfs_storage_api::{HTTPRangeSpec, ObjectIO, ObjectToDelete};
|
||||
use std::sync::Arc;
|
||||
use tokio_util::sync::CancellationToken;
|
||||
|
||||
pub(crate) const BUCKET_META_PREFIX: &str = rustfs_ecstore::disk::BUCKET_META_PREFIX;
|
||||
pub(crate) const RUSTFS_META_BUCKET: &str = rustfs_ecstore::disk::RUSTFS_META_BUCKET;
|
||||
pub(crate) const STORAGE_FORMAT_FILE: &str = rustfs_ecstore::disk::STORAGE_FORMAT_FILE;
|
||||
pub(crate) const TRANSITION_COMPLETE: &str = rustfs_ecstore::bucket::lifecycle::lifecycle::TRANSITION_COMPLETE;
|
||||
|
||||
pub(crate) type Disk = rustfs_ecstore::disk::Disk;
|
||||
pub(crate) type DiskError = rustfs_ecstore::disk::error::DiskError;
|
||||
pub(crate) type ECStore = rustfs_ecstore::store::ECStore;
|
||||
pub(crate) type EcstoreError = rustfs_ecstore::error::Error;
|
||||
pub(crate) type EcstoreResult<T> = rustfs_ecstore::error::Result<T>;
|
||||
pub(crate) type ListPathRawOptions = rustfs_ecstore::cache_value::metacache_set::ListPathRawOptions;
|
||||
pub(crate) type SetDisks = rustfs_ecstore::set_disk::SetDisks;
|
||||
pub(crate) type StorageError = rustfs_ecstore::error::StorageError;
|
||||
|
||||
pub(crate) use rustfs_ecstore::bucket::{
|
||||
bucket_target_sys::BucketTargetSys,
|
||||
lifecycle::{
|
||||
bucket_lifecycle_audit::LcEventSrc,
|
||||
bucket_lifecycle_ops::{GLOBAL_ExpiryState, apply_expiry_rule, apply_transition_rule},
|
||||
evaluator::Evaluator,
|
||||
lifecycle::{Event, Lifecycle, ObjectOpts},
|
||||
},
|
||||
metadata_sys::{get_lifecycle_config, get_object_lock_config, get_replication_config},
|
||||
replication::{ReplicationConfig, ReplicationConfigurationExt, ReplicationQueueAdmission, queue_replication_heal_internal},
|
||||
versioning::VersioningApi,
|
||||
versioning_sys::BucketVersioningSys,
|
||||
};
|
||||
pub(crate) use rustfs_ecstore::disk::{DiskAPI, DiskInfoOptions};
|
||||
pub(crate) use rustfs_ecstore::global::GLOBAL_TierConfigMgr;
|
||||
|
||||
pub(crate) mod storageclass {
|
||||
pub(crate) const RRS: &str = rustfs_ecstore::config::storageclass::RRS;
|
||||
pub(crate) const STANDARD: &str = rustfs_ecstore::config::storageclass::STANDARD;
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
pub(crate) use self::ecstore::{
|
||||
config::init as init_ecstore_config_for_scanner_tests,
|
||||
disk::{DiskOption, endpoint::Endpoint, new_disk},
|
||||
};
|
||||
pub(crate) use rustfs_ecstore::config::init as init_ecstore_config_for_scanner_tests;
|
||||
|
||||
#[cfg(test)]
|
||||
pub(crate) use rustfs_ecstore::disk::{DiskOption, endpoint::Endpoint, new_disk};
|
||||
|
||||
pub type ScannerGetObjectReader = rustfs_ecstore::store_api::GetObjectReader;
|
||||
pub type ScannerObjectInfo = rustfs_ecstore::store_api::ObjectInfo;
|
||||
@@ -65,7 +65,49 @@ pub type ScannerObjectToDelete = ObjectToDelete;
|
||||
pub type ScannerPutObjReader = rustfs_ecstore::store_api::PutObjReader;
|
||||
|
||||
pub(crate) fn resolve_scanner_object_store_handle() -> Option<Arc<ECStore>> {
|
||||
ecstore::resolve_object_store_handle()
|
||||
rustfs_ecstore::resolve_object_store_handle()
|
||||
}
|
||||
|
||||
pub(crate) fn is_reserved_or_invalid_bucket(bucket: &str, strict: bool) -> bool {
|
||||
rustfs_ecstore::store_utils::is_reserved_or_invalid_bucket(bucket, strict)
|
||||
}
|
||||
|
||||
pub(crate) fn path2_bucket_object(name: &str) -> (String, String) {
|
||||
rustfs_ecstore::pools::path2_bucket_object(name)
|
||||
}
|
||||
|
||||
pub(crate) fn path2_bucket_object_with_base_path(base_path: &str, path: &str) -> (String, String) {
|
||||
rustfs_ecstore::pools::path2_bucket_object_with_base_path(base_path, path)
|
||||
}
|
||||
|
||||
pub(crate) async fn is_erasure() -> bool {
|
||||
rustfs_ecstore::global::is_erasure().await
|
||||
}
|
||||
|
||||
pub(crate) async fn is_erasure_sd() -> bool {
|
||||
rustfs_ecstore::global::is_erasure_sd().await
|
||||
}
|
||||
|
||||
pub(crate) async fn read_config<S>(api: Arc<S>, file: &str) -> EcstoreResult<Vec<u8>>
|
||||
where
|
||||
S: ScannerObjectIO,
|
||||
{
|
||||
rustfs_ecstore::config::com::read_config(api, file).await
|
||||
}
|
||||
|
||||
pub(crate) async fn save_config<S>(api: Arc<S>, file: &str, data: Vec<u8>) -> EcstoreResult<()>
|
||||
where
|
||||
S: ScannerObjectIO,
|
||||
{
|
||||
rustfs_ecstore::config::com::save_config(api, file, data).await
|
||||
}
|
||||
|
||||
pub(crate) async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> std::result::Result<(), DiskError> {
|
||||
rustfs_ecstore::cache_value::metacache_set::list_path_raw(rx, opts).await
|
||||
}
|
||||
|
||||
pub(crate) async fn replace_bucket_usage_memory_from_info(data_usage_info: &rustfs_data_usage::DataUsageInfo) {
|
||||
rustfs_ecstore::data_usage::replace_bucket_usage_memory_from_info(data_usage_info).await;
|
||||
}
|
||||
|
||||
pub trait ScannerObjectIO:
|
||||
|
||||
Reference in New Issue
Block a user