refactor: remove external owner compat bridges (#3741)

This commit is contained in:
Zhengchao An
2026-06-22 18:28:35 +08:00
committed by GitHub
parent 539c68778d
commit d3796d6c10
25 changed files with 618 additions and 618 deletions
+1 -1
View File
@@ -14,7 +14,7 @@
use thiserror::Error;
use super::heal::storage_compat::{DiskError, EcstoreError};
use super::heal::{DiskError, EcstoreError};
/// Custom error type for heal operations
/// This enum defines various error variants that can occur during
+1 -1
View File
@@ -494,7 +494,7 @@ impl HealChannelProcessor {
#[cfg(test)]
mod tests {
use super::super::storage_compat::{DiskStore, Endpoint};
use super::super::{DiskStore, Endpoint};
use super::*;
use crate::heal::manager::HealConfig;
use crate::heal::storage::{HealObjectInfo, HealStorageAPI};
+1 -1
View File
@@ -28,7 +28,7 @@ use std::sync::{
use tokio::sync::{RwLock, Semaphore};
use tracing::{debug, error, info, warn};
use super::storage_compat::DiskStore;
use super::DiskStore;
const LOG_COMPONENT_HEAL: &str = "heal";
const LOG_SUBSYSTEM_ERASURE_HEALER: &str = "erasure_healer";
+1 -1
View File
@@ -17,7 +17,7 @@ use crate::{Error, Result};
use serde::{Deserialize, Serialize};
use std::time::SystemTime;
use super::storage_compat::Endpoint;
use super::Endpoint;
/// Corruption type
#[derive(Debug, Clone, Serialize, Deserialize)]
+2 -2
View File
@@ -33,7 +33,7 @@ use tokio::{
use tokio_util::sync::CancellationToken;
use tracing::{debug, error, info, warn};
use super::storage_compat::{DiskError, GLOBAL_LOCAL_DISK_MAP, HealDiskExt as _};
use super::{DiskError, GLOBAL_LOCAL_DISK_MAP, HealDiskExt as _};
const KEEP_HEAL_TASK_STATUS_DURATION: Duration = Duration::from_secs(10 * 60);
const LOG_COMPONENT_HEAL: &str = "heal";
@@ -2347,7 +2347,7 @@ mod tests {
use rustfs_madmin::heal_commands::HealResultItem;
use rustfs_storage_api::BucketInfo;
use super::super::storage_compat::{DiskStore, Endpoint};
use super::super::{DiskStore, Endpoint};
struct MockStorage;
+86 -1
View File
@@ -19,11 +19,96 @@ pub mod manager;
pub mod progress;
pub mod resume;
pub mod storage;
pub(crate) mod storage_compat;
pub mod task;
pub mod utils;
use rustfs_ecstore::api::data_usage as ecstore_data_usage;
use rustfs_ecstore::api::disk as ecstore_disk;
use rustfs_ecstore::api::error as ecstore_error;
use rustfs_ecstore::api::global as ecstore_global;
use rustfs_ecstore::api::storage as ecstore_storage;
pub use erasure_healer::ErasureSetHealer;
pub use manager::{HealManager, HealOperationsSnapshot, HealPriorityCounts, HealSourceCounts};
pub use resume::{CheckpointManager, ResumeCheckpoint, ResumeManager, ResumeState, ResumeUtils};
pub use task::{HealOptions, HealPriority, HealRequest, HealTask, HealType};
pub(crate) const DATA_USAGE_CACHE_NAME: &str = ecstore_data_usage::DATA_USAGE_CACHE_NAME;
pub(crate) const BUCKET_META_PREFIX: &str = ecstore_disk::BUCKET_META_PREFIX;
pub(crate) const RUSTFS_META_BUCKET: &str = ecstore_disk::RUSTFS_META_BUCKET;
pub(crate) type DiskError = ecstore_disk::error::DiskError;
pub(crate) type DiskResult<T> = ecstore_disk::error::Result<T>;
pub(crate) type DiskStore = ecstore_disk::DiskStore;
pub(crate) type ECStore = ecstore_storage::ECStore;
pub(crate) type EcstoreError = ecstore_error::Error;
pub(crate) type Endpoint = ecstore_disk::endpoint::Endpoint;
pub(crate) type StorageError = ecstore_error::StorageError;
pub(crate) type LocalDiskMap = std::collections::HashMap<String, Option<DiskStore>>;
pub(crate) struct GlobalLocalDiskMap;
pub(crate) static GLOBAL_LOCAL_DISK_MAP: GlobalLocalDiskMap = GlobalLocalDiskMap;
impl GlobalLocalDiskMap {
pub(crate) async fn read(&self) -> tokio::sync::RwLockReadGuard<'static, LocalDiskMap> {
ecstore_global::GLOBAL_LOCAL_DISK_MAP.read().await
}
}
#[cfg(test)]
pub(crate) type DiskOption = ecstore_disk::DiskOption;
#[cfg(test)]
pub(crate) async fn new_disk(ep: &Endpoint, opt: &DiskOption) -> ecstore_disk::error::Result<DiskStore> {
ecstore_disk::new_disk(ep, opt).await
}
pub(crate) trait HealDiskExt {
fn endpoint(&self) -> Endpoint;
async fn get_disk_id(&self) -> DiskResult<Option<uuid::Uuid>>;
async fn read_all(&self, volume: &str, path: &str) -> DiskResult<ecstore_disk::Bytes>;
async fn write_all(&self, volume: &str, path: &str, data: ecstore_disk::Bytes) -> DiskResult<()>;
async fn delete(&self, volume: &str, path: &str, options: ecstore_disk::DeleteOptions) -> DiskResult<()>;
async fn list_dir(&self, origvolume: &str, volume: &str, dir_path: &str, count: i32) -> DiskResult<Vec<String>>;
#[cfg(test)]
async fn make_volume(&self, volume: &str) -> DiskResult<()>;
}
impl<T> HealDiskExt for T
where
T: ecstore_disk::DiskAPI,
{
fn endpoint(&self) -> Endpoint {
ecstore_disk::DiskAPI::endpoint(self)
}
async fn get_disk_id(&self) -> DiskResult<Option<uuid::Uuid>> {
ecstore_disk::DiskAPI::get_disk_id(self).await
}
async fn read_all(&self, volume: &str, path: &str) -> DiskResult<ecstore_disk::Bytes> {
ecstore_disk::DiskAPI::read_all(self, volume, path).await
}
async fn write_all(&self, volume: &str, path: &str, data: ecstore_disk::Bytes) -> DiskResult<()> {
ecstore_disk::DiskAPI::write_all(self, volume, path, data).await
}
async fn delete(&self, volume: &str, path: &str, options: ecstore_disk::DeleteOptions) -> DiskResult<()> {
ecstore_disk::DiskAPI::delete(self, volume, path, options).await
}
async fn list_dir(&self, origvolume: &str, volume: &str, dir_path: &str, count: i32) -> DiskResult<Vec<String>> {
ecstore_disk::DiskAPI::list_dir(self, origvolume, volume, dir_path, count).await
}
#[cfg(test)]
async fn make_volume(&self, volume: &str) -> DiskResult<()> {
ecstore_disk::DiskAPI::make_volume(self, volume).await
}
}
pub type HealObjectInfo = <ECStore as rustfs_storage_api::ObjectOperations>::ObjectInfo;
pub type HealObjectOptions = <ECStore as rustfs_storage_api::ObjectOperations>::ObjectOptions;
pub type HealPutObjReader = <ECStore as rustfs_storage_api::ObjectIO>::PutObjectReader;
+2 -2
View File
@@ -21,7 +21,7 @@ use tokio::sync::RwLock;
use tracing::{debug, warn};
use uuid::Uuid;
use super::storage_compat::{BUCKET_META_PREFIX, DiskError, DiskStore, HealDiskExt as _, RUSTFS_META_BUCKET};
use super::{BUCKET_META_PREFIX, DiskError, DiskStore, HealDiskExt as _, RUSTFS_META_BUCKET};
const LOG_COMPONENT_HEAL: &str = "heal";
const LOG_SUBSYSTEM_RESUME: &str = "resume";
@@ -744,7 +744,7 @@ mod tests {
#[tokio::test]
async fn test_get_resumable_tasks_integration() {
use super::super::storage_compat::{DiskOption, Endpoint, new_disk};
use super::super::{DiskOption, Endpoint, new_disk};
use tempfile::TempDir;
// Create a temporary directory for testing
+3 -3
View File
@@ -23,8 +23,8 @@ use rustfs_storage_api::{
use std::sync::Arc;
use tracing::{debug, error, warn};
use super::storage_compat::{DiskStore, ECStore, Endpoint, StorageError};
pub use super::storage_compat::{HealObjectInfo, HealObjectOptions, HealPutObjReader};
use super::{DiskStore, ECStore, Endpoint, StorageError};
pub use super::{HealObjectInfo, HealObjectOptions, HealPutObjReader};
const LOG_COMPONENT_HEAL: &str = "heal";
const LOG_SUBSYSTEM_STORAGE: &str = "storage";
@@ -1232,7 +1232,7 @@ impl HealStorageAPI for ECStoreHealStorage {
#[cfg(test)]
mod tests {
use super::super::storage_compat::StorageError;
use super::super::StorageError;
use super::{is_transient_object_exists_error, is_transient_object_exists_message};
#[test]
-99
View File
@@ -1,99 +0,0 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use rustfs_ecstore::api::data_usage as ecstore_data_usage;
use rustfs_ecstore::api::disk as ecstore_disk;
use rustfs_ecstore::api::error as ecstore_error;
use rustfs_ecstore::api::global as ecstore_global;
use rustfs_ecstore::api::storage as ecstore_storage;
pub(crate) const DATA_USAGE_CACHE_NAME: &str = ecstore_data_usage::DATA_USAGE_CACHE_NAME;
pub(crate) const BUCKET_META_PREFIX: &str = ecstore_disk::BUCKET_META_PREFIX;
pub(crate) const RUSTFS_META_BUCKET: &str = ecstore_disk::RUSTFS_META_BUCKET;
pub(crate) type DiskError = ecstore_disk::error::DiskError;
pub(crate) type DiskResult<T> = ecstore_disk::error::Result<T>;
pub(crate) type DiskStore = ecstore_disk::DiskStore;
pub(crate) type ECStore = ecstore_storage::ECStore;
pub(crate) type EcstoreError = ecstore_error::Error;
pub(crate) type Endpoint = ecstore_disk::endpoint::Endpoint;
pub(crate) type StorageError = ecstore_error::StorageError;
pub(crate) type LocalDiskMap = std::collections::HashMap<String, Option<DiskStore>>;
pub(crate) struct GlobalLocalDiskMap;
pub(crate) static GLOBAL_LOCAL_DISK_MAP: GlobalLocalDiskMap = GlobalLocalDiskMap;
impl GlobalLocalDiskMap {
pub(crate) async fn read(&self) -> tokio::sync::RwLockReadGuard<'static, LocalDiskMap> {
ecstore_global::GLOBAL_LOCAL_DISK_MAP.read().await
}
}
#[cfg(test)]
pub(crate) type DiskOption = ecstore_disk::DiskOption;
#[cfg(test)]
pub(crate) async fn new_disk(ep: &Endpoint, opt: &DiskOption) -> ecstore_disk::error::Result<DiskStore> {
ecstore_disk::new_disk(ep, opt).await
}
pub(crate) trait HealDiskExt {
fn endpoint(&self) -> Endpoint;
async fn get_disk_id(&self) -> DiskResult<Option<uuid::Uuid>>;
async fn read_all(&self, volume: &str, path: &str) -> DiskResult<ecstore_disk::Bytes>;
async fn write_all(&self, volume: &str, path: &str, data: ecstore_disk::Bytes) -> DiskResult<()>;
async fn delete(&self, volume: &str, path: &str, options: ecstore_disk::DeleteOptions) -> DiskResult<()>;
async fn list_dir(&self, origvolume: &str, volume: &str, dir_path: &str, count: i32) -> DiskResult<Vec<String>>;
#[cfg(test)]
async fn make_volume(&self, volume: &str) -> DiskResult<()>;
}
impl<T> HealDiskExt for T
where
T: ecstore_disk::DiskAPI,
{
fn endpoint(&self) -> Endpoint {
ecstore_disk::DiskAPI::endpoint(self)
}
async fn get_disk_id(&self) -> DiskResult<Option<uuid::Uuid>> {
ecstore_disk::DiskAPI::get_disk_id(self).await
}
async fn read_all(&self, volume: &str, path: &str) -> DiskResult<ecstore_disk::Bytes> {
ecstore_disk::DiskAPI::read_all(self, volume, path).await
}
async fn write_all(&self, volume: &str, path: &str, data: ecstore_disk::Bytes) -> DiskResult<()> {
ecstore_disk::DiskAPI::write_all(self, volume, path, data).await
}
async fn delete(&self, volume: &str, path: &str, options: ecstore_disk::DeleteOptions) -> DiskResult<()> {
ecstore_disk::DiskAPI::delete(self, volume, path, options).await
}
async fn list_dir(&self, origvolume: &str, volume: &str, dir_path: &str, count: i32) -> DiskResult<Vec<String>> {
ecstore_disk::DiskAPI::list_dir(self, origvolume, volume, dir_path, count).await
}
#[cfg(test)]
async fn make_volume(&self, volume: &str) -> DiskResult<()> {
ecstore_disk::DiskAPI::make_volume(self, volume).await
}
}
pub type HealObjectInfo = <ECStore as rustfs_storage_api::ObjectOperations>::ObjectInfo;
pub type HealObjectOptions = <ECStore as rustfs_storage_api::ObjectOperations>::ObjectOptions;
pub type HealPutObjReader = <ECStore as rustfs_storage_api::ObjectIO>::PutObjectReader;
+2 -2
View File
@@ -27,7 +27,7 @@ use tokio::sync::RwLock;
use tracing::{debug, error, info, warn};
use uuid::Uuid;
use super::storage_compat::{BUCKET_META_PREFIX, DATA_USAGE_CACHE_NAME, RUSTFS_META_BUCKET};
use super::{BUCKET_META_PREFIX, DATA_USAGE_CACHE_NAME, RUSTFS_META_BUCKET};
const LOG_COMPONENT_HEAL: &str = "heal";
const LOG_SUBSYSTEM_TASK: &str = "task";
@@ -2047,7 +2047,7 @@ impl std::fmt::Debug for HealTask {
#[cfg(test)]
mod tests {
use super::super::storage_compat::{DiskStore, Endpoint};
use super::super::{DiskStore, Endpoint};
use super::*;
use crate::heal::storage::{DiskStatus, HealObjectInfo};
use rustfs_madmin::heal_commands::HealResultItem;
+1 -1
View File
@@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use super::storage_compat::IamStorageError;
use crate::IamStorageError;
use rustfs_policy::policy::Error as PolicyError;
pub type Result<T> = core::result::Result<T, Error>;
+155 -2
View File
@@ -12,10 +12,13 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use self::storage_compat::IamStore;
use crate::error::{Error, Result};
use manager::IamCache;
use oidc::OidcSys;
use rustfs_ecstore::api::{
config as ecstore_config, error as ecstore_error, global as ecstore_global, notification as ecstore_notification,
storage as ecstore_storage,
};
use std::sync::{Arc, OnceLock};
use store::object::ObjectStore;
use sys::IamSys;
@@ -33,11 +36,161 @@ pub mod keyring;
pub mod manager;
pub mod oidc;
pub mod oidc_state;
mod storage_compat;
pub mod store;
pub mod sys;
pub mod utils;
pub(crate) const IAM_CONFIG_ROOT_PREFIX: &str = ecstore_config::RUSTFS_CONFIG_PREFIX;
pub(crate) type IamEcstoreError = ecstore_error::Error;
pub(crate) type IamStorageError = ecstore_error::StorageError;
pub(crate) type IamStorageResult<T> = ecstore_error::Result<T>;
pub(crate) type IamStore = ecstore_storage::ECStore;
pub(crate) type IamConfigObjectInfo = <IamStore as rustfs_storage_api::ObjectOperations>::ObjectInfo;
pub(crate) type IamConfigObjectOptions = <IamStore as rustfs_storage_api::ObjectOperations>::ObjectOptions;
pub(crate) async fn read_iam_config_no_lock(api: Arc<IamStore>, file: &str) -> IamStorageResult<Vec<u8>> {
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)> {
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<()> {
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<()> {
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<()> {
ecstore_config::com::delete_config(api, file).await
}
pub(crate) fn classify_iam_system_path_failure_reason(err: &IamEcstoreError) -> &'static str {
ecstore_error::classify_system_path_failure_reason(err)
}
pub(crate) async fn is_iam_first_cluster_node_local() -> bool {
ecstore_global::is_first_cluster_node_local().await
}
pub(crate) struct IamNotificationPeerErr {
pub(crate) err: Option<IamEcstoreError>,
}
impl From<ecstore_notification::NotificationPeerErr> for IamNotificationPeerErr {
fn from(value: ecstore_notification::NotificationPeerErr) -> Self {
Self { err: value.err }
}
}
pub(crate) async fn notify_iam_delete_policy(policy_name: &str) -> Vec<IamNotificationPeerErr> {
match ecstore_notification::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 ecstore_notification::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 ecstore_notification::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 ecstore_notification::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 ecstore_notification::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 ecstore_notification::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 ecstore_notification::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 ecstore_notification::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(),
}
}
static IAM_SYS: OnceLock<Arc<IamSys<ObjectStore>>> = OnceLock::new();
static OIDC_SYS: OnceLock<Arc<OidcSys>> = OnceLock::new();
+1 -1
View File
@@ -12,8 +12,8 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use super::storage_compat::is_iam_first_cluster_node_local;
use crate::error::{Error, Result, is_err_config_not_found};
use crate::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},
-172
View File
@@ -1,172 +0,0 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use std::sync::Arc;
use rustfs_ecstore::api::config as ecstore_config;
use rustfs_ecstore::api::error as ecstore_error;
use rustfs_ecstore::api::global as ecstore_global;
use rustfs_ecstore::api::notification as ecstore_notification;
use rustfs_ecstore::api::storage as ecstore_storage;
pub(crate) const IAM_CONFIG_ROOT_PREFIX: &str = ecstore_config::RUSTFS_CONFIG_PREFIX;
pub(crate) type IamEcstoreError = ecstore_error::Error;
pub(crate) type IamStorageError = ecstore_error::StorageError;
pub(crate) type IamStorageResult<T> = ecstore_error::Result<T>;
pub(crate) type IamStore = ecstore_storage::ECStore;
pub(crate) type IamConfigObjectInfo = <IamStore as rustfs_storage_api::ObjectOperations>::ObjectInfo;
pub(crate) type IamConfigObjectOptions = <IamStore as rustfs_storage_api::ObjectOperations>::ObjectOptions;
pub(crate) async fn read_iam_config_no_lock(api: Arc<IamStore>, file: &str) -> IamStorageResult<Vec<u8>> {
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)> {
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<()> {
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<()> {
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<()> {
ecstore_config::com::delete_config(api, file).await
}
pub(crate) fn classify_iam_system_path_failure_reason(err: &IamEcstoreError) -> &'static str {
ecstore_error::classify_system_path_failure_reason(err)
}
pub(crate) async fn is_iam_first_cluster_node_local() -> bool {
ecstore_global::is_first_cluster_node_local().await
}
pub(crate) struct IamNotificationPeerErr {
pub(crate) err: Option<IamEcstoreError>,
}
impl From<ecstore_notification::NotificationPeerErr> for IamNotificationPeerErr {
fn from(value: ecstore_notification::NotificationPeerErr) -> Self {
Self { err: value.err }
}
}
pub(crate) async fn notify_iam_delete_policy(policy_name: &str) -> Vec<IamNotificationPeerErr> {
match ecstore_notification::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 ecstore_notification::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 ecstore_notification::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 ecstore_notification::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 ecstore_notification::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 ecstore_notification::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 ecstore_notification::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 ecstore_notification::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(),
}
}
+3 -3
View File
@@ -12,12 +12,12 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use super::super::storage_compat::{
use super::{GroupInfo, MappedPolicy, Store, UserType};
use crate::error::{Error, Result, is_err_config_not_found, is_err_no_such_group};
use crate::{
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 super::{GroupInfo, MappedPolicy, Store, UserType};
use crate::error::{Error, Result, is_err_config_not_found, is_err_no_such_group};
use crate::{
cache::{Cache, CacheEntity},
error::{is_err_no_such_policy, is_err_no_such_user},
+4 -4
View File
@@ -12,10 +12,6 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use super::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::error::Error as IamError;
use crate::error::is_err_no_such_account;
use crate::error::is_err_no_such_temp_account;
@@ -28,6 +24,10 @@ use crate::store::MappedPolicy;
use crate::store::Store;
use crate::store::UserType;
use crate::utils::{extract_claims, extract_claims_allow_missing_exp};
use crate::{
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 rustfs_credentials::{Credentials, EMBEDDED_POLICY_TYPE, INHERITED_POLICY_TYPE, get_global_action_cred};
use rustfs_madmin::AddOrUpdateUserReq;
use rustfs_madmin::GroupDesc;
+2 -4
View File
@@ -32,14 +32,12 @@ use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf};
use tokio::time::{Duration, Instant, sleep, timeout};
use tracing::warn;
use super::storage_compat::{
use crate::ScannerObjectIO;
use crate::{
BUCKET_META_PREFIX, EcstoreError as Error, EcstoreResult as StorageResult, RUSTFS_META_BUCKET, ReplicationConfig,
ScannerObjectInfo as ObjectInfo, ScannerObjectOptions as ObjectOptions, StorageError, TRANSITION_COMPLETE, save_config,
storageclass,
};
pub use super::storage_compat::{
ScannerGetObjectReader, ScannerObjectIO, ScannerObjectInfo, ScannerObjectOptions, ScannerObjectToDelete, ScannerPutObjReader,
};
// Data usage constants
pub const DATA_USAGE_ROOT: &str = SLASH_SEPARATOR;
+278 -1
View File
@@ -20,6 +20,23 @@
rust_2018_idioms
)]
use http::HeaderMap;
use rustfs_ecstore::api::bucket as ecstore_bucket;
use rustfs_ecstore::api::cache as ecstore_cache;
use rustfs_ecstore::api::capacity as ecstore_capacity;
use rustfs_ecstore::api::config as ecstore_config;
use rustfs_ecstore::api::data_usage as ecstore_data_usage;
use rustfs_ecstore::api::disk as ecstore_disk;
use rustfs_ecstore::api::error as ecstore_error;
use rustfs_ecstore::api::global as ecstore_global;
use rustfs_ecstore::api::set_disk as ecstore_set_disk;
use rustfs_ecstore::api::storage as ecstore_storage;
use rustfs_ecstore::api::tier as ecstore_tier;
use rustfs_storage_api::{HTTPRangeSpec, ObjectIO, ObjectToDelete};
use std::path::PathBuf;
use std::sync::Arc;
use tokio_util::sync::CancellationToken;
pub mod data_usage_define;
pub mod error;
pub mod runtime_config;
@@ -28,7 +45,6 @@ pub mod scanner_budget;
pub mod scanner_folder;
pub mod scanner_io;
pub mod sleeper;
mod storage_compat;
pub use data_usage_define::*;
pub use error::ScannerError;
@@ -60,3 +76,264 @@ impl Drop for ScannerActivityGuard {
.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| Some(current.saturating_sub(1)));
}
}
pub(crate) const BUCKET_META_PREFIX: &str = ecstore_disk::BUCKET_META_PREFIX;
pub(crate) const RUSTFS_META_BUCKET: &str = ecstore_disk::RUSTFS_META_BUCKET;
pub(crate) const STORAGE_FORMAT_FILE: &str = ecstore_disk::STORAGE_FORMAT_FILE;
pub(crate) const TRANSITION_COMPLETE: &str = ecstore_bucket::lifecycle::lifecycle::TRANSITION_COMPLETE;
pub(crate) type Disk = ecstore_disk::Disk;
#[cfg(test)]
pub(crate) type DiskStore = ecstore_disk::DiskStore;
pub(crate) type DiskLocation = ecstore_disk::DiskLocation;
pub(crate) type DiskError = ecstore_disk::error::DiskError;
pub(crate) type DiskResult<T> = ecstore_disk::error::Result<T>;
pub(crate) type ECStore = ecstore_storage::ECStore;
pub(crate) type EcstoreError = ecstore_error::Error;
pub(crate) type EcstoreResult<T> = ecstore_error::Result<T>;
pub(crate) type ListPathRawOptions = ecstore_cache::ListPathRawOptions;
pub(crate) type BucketTargetSys = ecstore_bucket::bucket_target_sys::BucketTargetSys;
pub(crate) type BucketVersioningSys = ecstore_bucket::versioning_sys::BucketVersioningSys;
pub(crate) type DiskInfoOptions = ecstore_disk::DiskInfoOptions;
pub(crate) type Evaluator = ecstore_bucket::lifecycle::evaluator::Evaluator;
pub(crate) type Event = ecstore_bucket::lifecycle::lifecycle::Event;
pub(crate) type LcEventSrc = ecstore_bucket::lifecycle::bucket_lifecycle_audit::LcEventSrc;
pub(crate) type ObjectOpts = ecstore_bucket::lifecycle::lifecycle::ObjectOpts;
pub(crate) type ReplicationConfig = ecstore_bucket::replication::ReplicationConfig;
pub(crate) type ReplicationHealQueueResult = ecstore_bucket::replication::ReplicationHealQueueResult;
pub(crate) type ReplicationQueueAdmission = ecstore_bucket::replication::ReplicationQueueAdmission;
pub(crate) type ScanGuard = ecstore_disk::ScanGuard;
pub(crate) type SetDisks = ecstore_set_disk::SetDisks;
pub(crate) type StorageError = ecstore_error::StorageError;
pub type ScannerGetObjectReader = <ECStore as ObjectIO>::GetObjectReader;
pub type ScannerObjectInfo = <ECStore as rustfs_storage_api::ObjectOperations>::ObjectInfo;
pub type ScannerObjectOptions = <ECStore as rustfs_storage_api::ObjectOperations>::ObjectOptions;
pub type ScannerObjectToDelete = ObjectToDelete;
pub type ScannerPutObjReader = <ECStore as ObjectIO>::PutObjectReader;
pub(crate) mod storageclass {
use super::ecstore_config;
pub(crate) const RRS: &str = ecstore_config::storageclass::RRS;
pub(crate) const STANDARD: &str = ecstore_config::storageclass::STANDARD;
}
#[cfg(test)]
pub(crate) fn init_ecstore_config_for_scanner_tests() {
ecstore_config::init();
}
#[cfg(test)]
pub(crate) type DiskOption = ecstore_disk::DiskOption;
#[cfg(test)]
pub(crate) type Endpoint = ecstore_disk::endpoint::Endpoint;
#[cfg(test)]
pub(crate) async fn new_disk(ep: &Endpoint, opt: &DiskOption) -> ecstore_disk::error::Result<DiskStore> {
ecstore_disk::new_disk(ep, opt).await
}
pub(crate) async fn get_lifecycle_config(
bucket: &str,
) -> EcstoreResult<(s3s::dto::BucketLifecycleConfiguration, time::OffsetDateTime)> {
ecstore_bucket::metadata_sys::get_lifecycle_config(bucket).await
}
pub(crate) async fn get_object_lock_config(
bucket: &str,
) -> EcstoreResult<(s3s::dto::ObjectLockConfiguration, time::OffsetDateTime)> {
ecstore_bucket::metadata_sys::get_object_lock_config(bucket).await
}
pub(crate) async fn get_replication_config(
bucket: &str,
) -> EcstoreResult<(s3s::dto::ReplicationConfiguration, time::OffsetDateTime)> {
ecstore_bucket::metadata_sys::get_replication_config(bucket).await
}
pub(crate) trait ScannerLifecycleConfigExt {
fn has_active_rules(&self, prefix: &str) -> bool;
}
impl ScannerLifecycleConfigExt for s3s::dto::BucketLifecycleConfiguration {
fn has_active_rules(&self, prefix: &str) -> bool {
<s3s::dto::BucketLifecycleConfiguration as ecstore_bucket::lifecycle::lifecycle::Lifecycle>::has_active_rules(
self, prefix,
)
}
}
pub(crate) trait ScannerReplicationConfigExt {
fn has_active_rules(&self, prefix: &str, recursive: bool) -> bool;
}
impl ScannerReplicationConfigExt for s3s::dto::ReplicationConfiguration {
fn has_active_rules(&self, prefix: &str, recursive: bool) -> bool {
<s3s::dto::ReplicationConfiguration as ecstore_bucket::replication::ReplicationConfigurationExt>::has_active_rules(
self, prefix, recursive,
)
}
}
pub(crate) trait ScannerVersioningConfigExt {
fn prefix_enabled(&self, prefix: &str) -> bool;
fn versioned(&self, prefix: &str) -> bool;
}
impl ScannerVersioningConfigExt for s3s::dto::VersioningConfiguration {
fn prefix_enabled(&self, prefix: &str) -> bool {
<s3s::dto::VersioningConfiguration as ecstore_bucket::versioning::VersioningApi>::prefix_enabled(self, prefix)
}
fn versioned(&self, prefix: &str) -> bool {
<s3s::dto::VersioningConfiguration as ecstore_bucket::versioning::VersioningApi>::versioned(self, prefix)
}
}
pub(crate) trait ScannerDiskExt {
async fn disk_info(&self, opts: &DiskInfoOptions) -> DiskResult<ecstore_disk::DiskInfo>;
async fn read_metadata(&self, volume: &str, path: &str) -> DiskResult<ecstore_disk::Bytes>;
fn path(&self) -> PathBuf;
fn get_disk_location(&self) -> DiskLocation;
fn start_scan(&self) -> ScanGuard;
}
impl<T> ScannerDiskExt for T
where
T: ecstore_disk::DiskAPI,
{
async fn disk_info(&self, opts: &DiskInfoOptions) -> DiskResult<ecstore_disk::DiskInfo> {
ecstore_disk::DiskAPI::disk_info(self, opts).await
}
async fn read_metadata(&self, volume: &str, path: &str) -> DiskResult<ecstore_disk::Bytes> {
ecstore_disk::DiskAPI::read_metadata(self, volume, path).await
}
fn path(&self) -> PathBuf {
ecstore_disk::DiskAPI::path(self)
}
fn get_disk_location(&self) -> DiskLocation {
ecstore_disk::DiskAPI::get_disk_location(self)
}
fn start_scan(&self) -> ScanGuard {
ecstore_disk::DiskAPI::start_scan(self)
}
}
pub(crate) async fn apply_transition_rule(event: &Event, src: &LcEventSrc, oi: &ScannerObjectInfo) -> bool {
ecstore_bucket::lifecycle::bucket_lifecycle_ops::apply_transition_rule(event, src, oi).await
}
pub(crate) async fn apply_expiry_rule(event: &Event, src: &LcEventSrc, oi: &ScannerObjectInfo) -> bool {
ecstore_bucket::lifecycle::bucket_lifecycle_ops::apply_expiry_rule(event, src, oi).await
}
pub(crate) async fn list_global_tiers() -> Vec<ecstore_tier::tier_config::TierConfig> {
ecstore_global::GLOBAL_TierConfigMgr.read().await.list_tiers()
}
pub(crate) async fn enqueue_global_free_version(oi: ScannerObjectInfo) {
ecstore_bucket::lifecycle::bucket_lifecycle_ops::GLOBAL_ExpiryState
.write()
.await
.enqueue_free_version(oi)
.await;
}
pub(crate) async fn enqueue_global_newer_noncurrent(
bucket: &str,
to_delete_objs: Vec<ObjectToDelete>,
event: Event,
src: &LcEventSrc,
) -> bool {
ecstore_bucket::lifecycle::bucket_lifecycle_ops::GLOBAL_ExpiryState
.write()
.await
.enqueue_by_newer_noncurrent(bucket, to_delete_objs, event, src)
.await
}
pub(crate) async fn queue_replication_heal_internal(
bucket: &str,
oi: ScannerObjectInfo,
rcfg: ReplicationConfig,
retry_count: u32,
) -> ReplicationHealQueueResult {
ecstore_bucket::replication::queue_replication_heal_internal(bucket, oi, rcfg, retry_count).await
}
pub(crate) fn resolve_scanner_object_store_handle() -> Option<Arc<ECStore>> {
ecstore_global::resolve_object_store_handle()
}
pub(crate) fn is_reserved_or_invalid_bucket(bucket: &str, strict: bool) -> bool {
ecstore_capacity::is_reserved_or_invalid_bucket(bucket, strict)
}
pub(crate) fn path2_bucket_object(name: &str) -> (String, String) {
ecstore_capacity::path2_bucket_object(name)
}
pub(crate) fn path2_bucket_object_with_base_path(base_path: &str, path: &str) -> (String, String) {
ecstore_capacity::path2_bucket_object_with_base_path(base_path, path)
}
pub(crate) async fn is_erasure() -> bool {
ecstore_global::is_erasure().await
}
pub(crate) async fn is_erasure_sd() -> bool {
ecstore_global::is_erasure_sd().await
}
pub(crate) async fn read_config<S>(api: Arc<S>, file: &str) -> EcstoreResult<Vec<u8>>
where
S: ScannerObjectIO,
{
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,
{
ecstore_config::com::save_config(api, file, data).await
}
pub(crate) async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> std::result::Result<(), DiskError> {
ecstore_cache::list_path_raw(rx, opts).await
}
pub(crate) async fn replace_bucket_usage_memory_from_info(data_usage_info: &rustfs_data_usage::DataUsageInfo) {
ecstore_data_usage::replace_bucket_usage_memory_from_info(data_usage_info).await;
}
pub trait ScannerObjectIO:
ObjectIO<
Error = EcstoreError,
RangeSpec = HTTPRangeSpec,
HeaderMap = HeaderMap,
ObjectOptions = ScannerObjectOptions,
ObjectInfo = ScannerObjectInfo,
GetObjectReader = ScannerGetObjectReader,
PutObjectReader = ScannerPutObjReader,
>
{
}
impl<T> ScannerObjectIO for T where
T: ObjectIO<
Error = EcstoreError,
RangeSpec = HTTPRangeSpec,
HeaderMap = HeaderMap,
ObjectOptions = ScannerObjectOptions,
ObjectInfo = ScannerObjectInfo,
GetObjectReader = ScannerGetObjectReader,
PutObjectReader = ScannerPutObjReader,
>
{
}
+1 -1
View File
@@ -816,8 +816,8 @@ pub(crate) fn scanner_alert_excess_folders() -> u64 {
#[cfg(test)]
mod tests {
use super::super::storage_compat::init_ecstore_config_for_scanner_tests;
use super::{ScannerRuntimeConfigSource, lookup_scanner_runtime_config, validate_scanner_runtime_config};
use crate::init_ecstore_config_for_scanner_tests;
use rustfs_config::server_config::{Config as ServerConfig, KVS};
use rustfs_config::{
DEFAULT_DELIMITER, ENV_SCANNER_BITROT_CYCLE_SECS, ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS, ENV_SCANNER_CYCLE,
+4 -5
View File
@@ -14,9 +14,8 @@
use std::sync::Arc;
use crate::data_usage_define::{
BACKGROUND_HEAL_INFO_PATH, DATA_USAGE_BLOOM_NAME_PATH, DATA_USAGE_OBJ_NAME_PATH, ScannerObjectIO,
};
use crate::ScannerObjectIO;
use crate::data_usage_define::{BACKGROUND_HEAL_INFO_PATH, DATA_USAGE_BLOOM_NAME_PATH, DATA_USAGE_OBJ_NAME_PATH};
use crate::runtime_config::{
current_scanner_runtime_config, lookup_scanner_runtime_config, refresh_scanner_runtime_config_from_global,
scanner_bitrot_cycle, scanner_cycle_interval, scanner_start_delay, set_scanner_default_cycle_secs,
@@ -46,7 +45,7 @@ use tokio::time::{Duration, Instant};
use tokio_util::sync::CancellationToken;
use tracing::{debug, error, info, instrument, warn};
use super::storage_compat::{
use crate::{
ECStore, EcstoreError, RUSTFS_META_BUCKET, ScannerLifecycleConfigExt as _, ScannerReplicationConfigExt as _,
get_lifecycle_config, get_replication_config, is_erasure_sd, read_config, replace_bucket_usage_memory_from_info, save_config,
};
@@ -1104,8 +1103,8 @@ pub async fn store_data_usage_in_backend(
#[cfg(test)]
mod tests {
use super::super::storage_compat::EcstoreResult;
use super::*;
use crate::EcstoreResult;
use crate::{
ScannerGetObjectReader as GetObjectReader, ScannerObjectInfo as ObjectInfo, ScannerObjectOptions as ObjectOptions,
ScannerPutObjReader as PutObjReader,
+2 -2
View File
@@ -50,7 +50,7 @@ use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;
use tracing::{debug, error, warn};
use super::storage_compat::{
use crate::{
BucketVersioningSys, Disk, DiskError, DiskInfoOptions, Evaluator, Event, LcEventSrc, ListPathRawOptions, ObjectOpts,
ReplicationConfig, ReplicationQueueAdmission, ScannerDiskExt as _, ScannerLifecycleConfigExt as _,
ScannerReplicationConfigExt as _, ScannerVersioningConfigExt as _, StorageError, apply_expiry_rule, apply_transition_rule,
@@ -2419,8 +2419,8 @@ pub async fn scan_data_folder(
mod tests {
use crate::SCANNER_SLEEPER;
use super::super::storage_compat::{DiskOption, Endpoint, new_disk};
use super::*;
use crate::{DiskOption, Endpoint, new_disk};
use rustfs_filemeta::{ReplicateObjectInfo, ReplicationType, ResyncDecision, ResyncTargetDecision, VersionPurgeStatusType};
use serial_test::serial;
#[cfg(unix)]
+3 -3
View File
@@ -42,14 +42,14 @@ use tokio::time::Duration;
use tokio_util::sync::CancellationToken;
use tracing::{debug, error, warn};
use super::storage_compat::{
use crate::ScannerObjectInfo as ObjectInfo;
use crate::{
BucketTargetSys, BucketVersioningSys, Disk, DiskError, ECStore, EcstoreError as Error, EcstoreResult as Result,
ReplicationConfig, STORAGE_FORMAT_FILE, ScannerDiskExt as _, ScannerLifecycleConfigExt as _,
ScannerReplicationConfigExt as _, ScannerVersioningConfigExt as _, SetDisks, StorageError, enqueue_global_free_version,
get_lifecycle_config, get_object_lock_config, get_replication_config, list_global_tiers, resolve_scanner_object_store_handle,
storageclass,
};
use crate::ScannerObjectInfo as ObjectInfo;
pub(crate) const SCANNER_SKIP_FILE_ERROR: &str = "skip file";
const LOG_COMPONENT_SCANNER: &str = "scanner";
@@ -1522,9 +1522,9 @@ impl ScannerIODisk for Disk {
#[cfg(test)]
mod tests {
use super::super::storage_compat::{DiskOption, Endpoint, new_disk, path2_bucket_object_with_base_path};
use super::*;
use crate::scanner_folder::ScannerItem;
use crate::{DiskOption, Endpoint, new_disk, path2_bucket_object_with_base_path};
use serial_test::serial;
use temp_env::with_var;
use uuid::Uuid;
-291
View File
@@ -1,291 +0,0 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use http::HeaderMap;
use rustfs_ecstore::api::bucket as ecstore_bucket;
use rustfs_ecstore::api::cache as ecstore_cache;
use rustfs_ecstore::api::capacity as ecstore_capacity;
use rustfs_ecstore::api::config as ecstore_config;
use rustfs_ecstore::api::data_usage as ecstore_data_usage;
use rustfs_ecstore::api::disk as ecstore_disk;
use rustfs_ecstore::api::error as ecstore_error;
use rustfs_ecstore::api::global as ecstore_global;
use rustfs_ecstore::api::set_disk as ecstore_set_disk;
use rustfs_ecstore::api::storage as ecstore_storage;
use rustfs_ecstore::api::tier as ecstore_tier;
use rustfs_storage_api::{HTTPRangeSpec, ObjectIO, ObjectToDelete};
use std::path::PathBuf;
use std::sync::Arc;
use tokio_util::sync::CancellationToken;
pub(crate) const BUCKET_META_PREFIX: &str = ecstore_disk::BUCKET_META_PREFIX;
pub(crate) const RUSTFS_META_BUCKET: &str = ecstore_disk::RUSTFS_META_BUCKET;
pub(crate) const STORAGE_FORMAT_FILE: &str = ecstore_disk::STORAGE_FORMAT_FILE;
pub(crate) const TRANSITION_COMPLETE: &str = ecstore_bucket::lifecycle::lifecycle::TRANSITION_COMPLETE;
pub(crate) type Disk = ecstore_disk::Disk;
#[cfg(test)]
pub(crate) type DiskStore = ecstore_disk::DiskStore;
pub(crate) type DiskLocation = ecstore_disk::DiskLocation;
pub(crate) type DiskError = ecstore_disk::error::DiskError;
pub(crate) type DiskResult<T> = ecstore_disk::error::Result<T>;
pub(crate) type ECStore = ecstore_storage::ECStore;
pub(crate) type EcstoreError = ecstore_error::Error;
pub(crate) type EcstoreResult<T> = ecstore_error::Result<T>;
pub(crate) type ListPathRawOptions = ecstore_cache::ListPathRawOptions;
pub(crate) type BucketTargetSys = ecstore_bucket::bucket_target_sys::BucketTargetSys;
pub(crate) type BucketVersioningSys = ecstore_bucket::versioning_sys::BucketVersioningSys;
pub(crate) type DiskInfoOptions = ecstore_disk::DiskInfoOptions;
pub(crate) type Evaluator = ecstore_bucket::lifecycle::evaluator::Evaluator;
pub(crate) type Event = ecstore_bucket::lifecycle::lifecycle::Event;
pub(crate) type LcEventSrc = ecstore_bucket::lifecycle::bucket_lifecycle_audit::LcEventSrc;
pub(crate) type ObjectOpts = ecstore_bucket::lifecycle::lifecycle::ObjectOpts;
pub(crate) type ReplicationConfig = ecstore_bucket::replication::ReplicationConfig;
pub(crate) type ReplicationHealQueueResult = ecstore_bucket::replication::ReplicationHealQueueResult;
pub(crate) type ReplicationQueueAdmission = ecstore_bucket::replication::ReplicationQueueAdmission;
pub(crate) type ScanGuard = ecstore_disk::ScanGuard;
pub(crate) type SetDisks = ecstore_set_disk::SetDisks;
pub(crate) type StorageError = ecstore_error::StorageError;
pub type ScannerGetObjectReader = <ECStore as ObjectIO>::GetObjectReader;
pub type ScannerObjectInfo = <ECStore as rustfs_storage_api::ObjectOperations>::ObjectInfo;
pub type ScannerObjectOptions = <ECStore as rustfs_storage_api::ObjectOperations>::ObjectOptions;
pub type ScannerObjectToDelete = ObjectToDelete;
pub type ScannerPutObjReader = <ECStore as ObjectIO>::PutObjectReader;
pub(crate) mod storageclass {
use super::ecstore_config;
pub(crate) const RRS: &str = ecstore_config::storageclass::RRS;
pub(crate) const STANDARD: &str = ecstore_config::storageclass::STANDARD;
}
#[cfg(test)]
pub(crate) fn init_ecstore_config_for_scanner_tests() {
ecstore_config::init();
}
#[cfg(test)]
pub(crate) type DiskOption = ecstore_disk::DiskOption;
#[cfg(test)]
pub(crate) type Endpoint = ecstore_disk::endpoint::Endpoint;
#[cfg(test)]
pub(crate) async fn new_disk(ep: &Endpoint, opt: &DiskOption) -> ecstore_disk::error::Result<DiskStore> {
ecstore_disk::new_disk(ep, opt).await
}
pub(crate) async fn get_lifecycle_config(
bucket: &str,
) -> EcstoreResult<(s3s::dto::BucketLifecycleConfiguration, time::OffsetDateTime)> {
ecstore_bucket::metadata_sys::get_lifecycle_config(bucket).await
}
pub(crate) async fn get_object_lock_config(
bucket: &str,
) -> EcstoreResult<(s3s::dto::ObjectLockConfiguration, time::OffsetDateTime)> {
ecstore_bucket::metadata_sys::get_object_lock_config(bucket).await
}
pub(crate) async fn get_replication_config(
bucket: &str,
) -> EcstoreResult<(s3s::dto::ReplicationConfiguration, time::OffsetDateTime)> {
ecstore_bucket::metadata_sys::get_replication_config(bucket).await
}
pub(crate) trait ScannerLifecycleConfigExt {
fn has_active_rules(&self, prefix: &str) -> bool;
}
impl ScannerLifecycleConfigExt for s3s::dto::BucketLifecycleConfiguration {
fn has_active_rules(&self, prefix: &str) -> bool {
<s3s::dto::BucketLifecycleConfiguration as ecstore_bucket::lifecycle::lifecycle::Lifecycle>::has_active_rules(
self, prefix,
)
}
}
pub(crate) trait ScannerReplicationConfigExt {
fn has_active_rules(&self, prefix: &str, recursive: bool) -> bool;
}
impl ScannerReplicationConfigExt for s3s::dto::ReplicationConfiguration {
fn has_active_rules(&self, prefix: &str, recursive: bool) -> bool {
<s3s::dto::ReplicationConfiguration as ecstore_bucket::replication::ReplicationConfigurationExt>::has_active_rules(
self, prefix, recursive,
)
}
}
pub(crate) trait ScannerVersioningConfigExt {
fn prefix_enabled(&self, prefix: &str) -> bool;
fn versioned(&self, prefix: &str) -> bool;
}
impl ScannerVersioningConfigExt for s3s::dto::VersioningConfiguration {
fn prefix_enabled(&self, prefix: &str) -> bool {
<s3s::dto::VersioningConfiguration as ecstore_bucket::versioning::VersioningApi>::prefix_enabled(self, prefix)
}
fn versioned(&self, prefix: &str) -> bool {
<s3s::dto::VersioningConfiguration as ecstore_bucket::versioning::VersioningApi>::versioned(self, prefix)
}
}
pub(crate) trait ScannerDiskExt {
async fn disk_info(&self, opts: &DiskInfoOptions) -> DiskResult<ecstore_disk::DiskInfo>;
async fn read_metadata(&self, volume: &str, path: &str) -> DiskResult<ecstore_disk::Bytes>;
fn path(&self) -> PathBuf;
fn get_disk_location(&self) -> DiskLocation;
fn start_scan(&self) -> ScanGuard;
}
impl<T> ScannerDiskExt for T
where
T: ecstore_disk::DiskAPI,
{
async fn disk_info(&self, opts: &DiskInfoOptions) -> DiskResult<ecstore_disk::DiskInfo> {
ecstore_disk::DiskAPI::disk_info(self, opts).await
}
async fn read_metadata(&self, volume: &str, path: &str) -> DiskResult<ecstore_disk::Bytes> {
ecstore_disk::DiskAPI::read_metadata(self, volume, path).await
}
fn path(&self) -> PathBuf {
ecstore_disk::DiskAPI::path(self)
}
fn get_disk_location(&self) -> DiskLocation {
ecstore_disk::DiskAPI::get_disk_location(self)
}
fn start_scan(&self) -> ScanGuard {
ecstore_disk::DiskAPI::start_scan(self)
}
}
pub(crate) async fn apply_transition_rule(event: &Event, src: &LcEventSrc, oi: &ScannerObjectInfo) -> bool {
ecstore_bucket::lifecycle::bucket_lifecycle_ops::apply_transition_rule(event, src, oi).await
}
pub(crate) async fn apply_expiry_rule(event: &Event, src: &LcEventSrc, oi: &ScannerObjectInfo) -> bool {
ecstore_bucket::lifecycle::bucket_lifecycle_ops::apply_expiry_rule(event, src, oi).await
}
pub(crate) async fn list_global_tiers() -> Vec<ecstore_tier::tier_config::TierConfig> {
ecstore_global::GLOBAL_TierConfigMgr.read().await.list_tiers()
}
pub(crate) async fn enqueue_global_free_version(oi: ScannerObjectInfo) {
ecstore_bucket::lifecycle::bucket_lifecycle_ops::GLOBAL_ExpiryState
.write()
.await
.enqueue_free_version(oi)
.await;
}
pub(crate) async fn enqueue_global_newer_noncurrent(
bucket: &str,
to_delete_objs: Vec<ObjectToDelete>,
event: Event,
src: &LcEventSrc,
) -> bool {
ecstore_bucket::lifecycle::bucket_lifecycle_ops::GLOBAL_ExpiryState
.write()
.await
.enqueue_by_newer_noncurrent(bucket, to_delete_objs, event, src)
.await
}
pub(crate) async fn queue_replication_heal_internal(
bucket: &str,
oi: ScannerObjectInfo,
rcfg: ReplicationConfig,
retry_count: u32,
) -> ReplicationHealQueueResult {
ecstore_bucket::replication::queue_replication_heal_internal(bucket, oi, rcfg, retry_count).await
}
pub(crate) fn resolve_scanner_object_store_handle() -> Option<Arc<ECStore>> {
ecstore_global::resolve_object_store_handle()
}
pub(crate) fn is_reserved_or_invalid_bucket(bucket: &str, strict: bool) -> bool {
ecstore_capacity::is_reserved_or_invalid_bucket(bucket, strict)
}
pub(crate) fn path2_bucket_object(name: &str) -> (String, String) {
ecstore_capacity::path2_bucket_object(name)
}
pub(crate) fn path2_bucket_object_with_base_path(base_path: &str, path: &str) -> (String, String) {
ecstore_capacity::path2_bucket_object_with_base_path(base_path, path)
}
pub(crate) async fn is_erasure() -> bool {
ecstore_global::is_erasure().await
}
pub(crate) async fn is_erasure_sd() -> bool {
ecstore_global::is_erasure_sd().await
}
pub(crate) async fn read_config<S>(api: Arc<S>, file: &str) -> EcstoreResult<Vec<u8>>
where
S: ScannerObjectIO,
{
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,
{
ecstore_config::com::save_config(api, file, data).await
}
pub(crate) async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> std::result::Result<(), DiskError> {
ecstore_cache::list_path_raw(rx, opts).await
}
pub(crate) async fn replace_bucket_usage_memory_from_info(data_usage_info: &rustfs_data_usage::DataUsageInfo) {
ecstore_data_usage::replace_bucket_usage_memory_from_info(data_usage_info).await;
}
pub trait ScannerObjectIO:
ObjectIO<
Error = EcstoreError,
RangeSpec = HTTPRangeSpec,
HeaderMap = HeaderMap,
ObjectOptions = ScannerObjectOptions,
ObjectInfo = ScannerObjectInfo,
GetObjectReader = ScannerGetObjectReader,
PutObjectReader = ScannerPutObjReader,
>
{
}
impl<T> ScannerObjectIO for T where
T: ObjectIO<
Error = EcstoreError,
RangeSpec = HTTPRangeSpec,
HeaderMap = HeaderMap,
ObjectOptions = ScannerObjectOptions,
ObjectInfo = ScannerObjectInfo,
GetObjectReader = ScannerGetObjectReader,
PutObjectReader = ScannerPutObjReader,
>
{
}
+38 -12
View File
@@ -5,17 +5,17 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block
## Current Context
- Issue: [`rustfs/backlog#660`](https://github.com/rustfs/backlog/issues/660)
- Branch: `overtrue/arch-standalone-thin-compat-cleanup`
- Baseline: completed `C-011/C-012/C-013/API-055/API-059/API-079/API-080/API-081/API-082/API-083/API-084/API-085/API-086/API-087/API-088/API-089/API-090/API-091/API-092/API-093/API-094/API-095/API-096/API-097/API-098/API-099/API-100/API-101/API-102/API-103/API-104/API-105/API-106/API-107/API-108/API-109/API-110/API-111/API-112/API-113/API-114/API-115/API-116/API-117/API-118/API-119/API-120/API-121/API-122/API-123/API-124`.
- Based on: API-124 slice.
- Branch: `overtrue/arch-external-owner-compat-cleanup`
- Baseline: completed `C-011/C-012/C-013/API-055/API-059/API-079/API-080/API-081/API-082/API-083/API-084/API-085/API-086/API-087/API-088/API-089/API-090/API-091/API-092/API-093/API-094/API-095/API-096/API-097/API-098/API-099/API-100/API-101/API-102/API-103/API-104/API-105/API-106/API-107/API-108/API-109/API-110/API-111/API-112/API-113/API-114/API-115/API-116/API-117/API-118/API-119/API-120/API-121/API-122/API-123/API-124/API-125/API-126`.
- Based on: API-126 slice.
- PR type for this branch: `pure-move`
- Runtime behavior changes: none.
- Rust code changes: remove standalone thin storage compatibility bridges from
e2e, IAM store, notify, OBS, Swift, and S3 Select, then route their consumers
directly to ECStore API owner modules.
- CI/script changes: allow migrated standalone consumers to import owner APIs
directly and reject reintroduced bridge modules.
- Docs changes: record the API-125/API-126 standalone bridge cleanup.
- Rust code changes: remove external owner storage compatibility bridges from
IAM root, heal, and scanner, then route their consumers directly through
owner API definitions.
- CI/script changes: allow migrated owner roots to import ECStore facade APIs
directly and reject reintroduced IAM/heal/scanner bridge modules.
- Docs changes: record the API-127 external owner bridge cleanup.
## Phase 0 Tasks
@@ -778,6 +778,19 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block
- Verification: affected crate compile coverage, remaining standalone bridge
residual scan, migration and layer guards, formatting, diff hygiene, Rust
risk scan, and three-expert review.
- [x] `API-127` Remove external owner compatibility bridges.
- Completed slice: move IAM root, heal, and scanner bridge contracts into
their owner modules, delete their local `storage_compat.rs` bridge modules,
and update consumers to import owner APIs directly.
- Acceptance: IAM root, heal, and scanner no longer route through local
storage compatibility bridges; migration rules reject deleted files, module
declarations, or bridge consumers.
- Must preserve: IAM config object IO and notification wrappers, heal disk
extension behavior and object aliases, scanner lifecycle/replication/disk
wrappers, data-usage persistence, and scanner object IO contracts.
- Verification: focused IAM/heal/scanner compile coverage, external owner
bridge residual scan, migration and layer guards, formatting, diff hygiene,
Rust risk scan, and three-expert review.
- [x] `G-012` Inventory placement and repair invariants.
- Acceptance:
[`placement-repair-invariants.md`](placement-repair-invariants.md) records
@@ -3811,14 +3824,27 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block
| Expert | Status | Notes |
|---|---|---|
| Quality/architecture | pass | API-125/API-126 remove only standalone thin bridge modules and keep call sites on explicit owner APIs. |
| Migration preservation | pass | The migration guard now rejects the deleted e2e, IAM-store, notify, OBS, Swift, and S3 Select bridge files, module declarations, and bridge consumers. |
| Testing/verification | pass | Affected crate compile, residual scan, migration guard, layer guard, formatting, diff hygiene, and diff-only Rust risk scan passed. |
| Quality/architecture | pass | API-127 removes only external owner bridge modules and keeps call sites on explicit owner APIs. |
| Migration preservation | pass | The migration guard now rejects deleted IAM-root, heal, and scanner bridge files, module declarations, and bridge consumers. |
| Testing/verification | pass | Focused IAM/heal/scanner compile, residual scan, migration guard, layer guard, formatting, diff hygiene, and diff-only Rust risk scan passed. |
## Verification Notes
Passed before push:
- Issue #660 API-127 current slice:
- `cargo check --tests -p rustfs-iam -p rustfs-heal -p rustfs-scanner`: passed.
- `cargo fmt --all`: passed.
- `cargo fmt --all --check`: passed.
- `git diff --check`: passed.
- `bash -n scripts/check_architecture_migration_rules.sh`: passed.
- `./scripts/check_architecture_migration_rules.sh`: passed.
- `./scripts/check_layer_dependencies.sh`: passed.
- External owner compatibility bridge residual scan: passed.
- Rust risk scan: diff-only scan found no new unwrap/expect, numeric casts,
string-error public APIs, boxed public errors, println/eprintln, or relaxed
ordering.
- Issue #660 API-126 current slice:
- `cargo check --tests -p e2e_test -p rustfs-iam -p rustfs-notify -p rustfs-obs -p rustfs-protocols -p rustfs-s3select-api`: passed.
- `cargo fmt --all`: passed.
+27 -3
View File
@@ -120,6 +120,7 @@ E2E_STORAGE_COMPAT_RPC_PASSTHROUGH_HITS_FILE="${TMP_DIR}/e2e_storage_compat_rpc_
TEST_STORAGE_COMPAT_PASSTHROUGH_HITS_FILE="${TMP_DIR}/test_storage_compat_passthrough_hits.txt"
TEST_FUZZ_COMPAT_BRIDGE_HITS_FILE="${TMP_DIR}/test_fuzz_compat_bridge_hits.txt"
STANDALONE_THIN_COMPAT_BRIDGE_HITS_FILE="${TMP_DIR}/standalone_thin_compat_bridge_hits.txt"
EXTERNAL_OWNER_COMPAT_BRIDGE_HITS_FILE="${TMP_DIR}/external_owner_compat_bridge_hits.txt"
PRODUCTION_UNUSED_COMPAT_ALLOW_HITS_FILE="${TMP_DIR}/production_unused_compat_allow_hits.txt"
BROAD_STORE_API_COMPAT_REEXPORT_HITS_FILE="${TMP_DIR}/broad_store_api_compat_reexport_hits.txt"
NESTED_STORE_API_COMPAT_MODULE_HITS_FILE="${TMP_DIR}/nested_store_api_compat_module_hits.txt"
@@ -714,7 +715,7 @@ fi
--glob '!**/storage_compat.rs' \
--glob '!**/*storage_compat.rs' \
--glob '!target/**' \
| rg -v '^(rustfs/src/(capacity/service|config/config_test|error|init|runtime_capabilities|startup_(background|bucket_metadata|fs_guard|iam|lifecycle|notification|server|services|shutdown|storage)|table_catalog|workload_admission)\.rs|rustfs/src/server/(event|http|module_switch|readiness)\.rs|rustfs/src/storage/s3_api/(bucket|multipart)\.rs|crates/e2e_test/src/(replication_extension_test|reliant/(grpc_lock_client|node_interact_test))\.rs|crates/heal/tests/(endpoint_index_test|heal_bug_fixes_test|heal_integration_test)\.rs|crates/notify/src/config_manager\.rs|crates/obs/src/metrics/(scheduler|stats_collector)\.rs|crates/protocols/src/swift/mod\.rs|crates/s3select-api/src/object_store\.rs|crates/scanner/tests/lifecycle_integration_test\.rs|fuzz/fuzz_targets/(bucket_validation|path_containment)\.rs):' || true
| rg -v '^(rustfs/src/(capacity/service|config/config_test|error|init|runtime_capabilities|startup_(background|bucket_metadata|fs_guard|iam|lifecycle|notification|server|services|shutdown|storage)|table_catalog|workload_admission)\.rs|rustfs/src/server/(event|http|module_switch|readiness)\.rs|rustfs/src/storage/s3_api/(bucket|multipart)\.rs|crates/e2e_test/src/(replication_extension_test|reliant/(grpc_lock_client|node_interact_test))\.rs|crates/heal/src/heal/mod\.rs|crates/heal/tests/(endpoint_index_test|heal_bug_fixes_test|heal_integration_test)\.rs|crates/iam/src/lib\.rs|crates/notify/src/config_manager\.rs|crates/obs/src/metrics/(scheduler|stats_collector)\.rs|crates/protocols/src/swift/mod\.rs|crates/s3select-api/src/object_store\.rs|crates/scanner/src/lib\.rs|crates/scanner/tests/lifecycle_integration_test\.rs|fuzz/fuzz_targets/(bucket_validation|path_containment)\.rs):' || true
) |
cat >"$DIRECT_ECSTORE_IMPORT_HITS_FILE"
@@ -1339,6 +1340,27 @@ if [[ -s "$STANDALONE_THIN_COMPAT_BRIDGE_HITS_FILE" ]]; then
report_failure "standalone e2e/IAM-store/notify consumers must import owner APIs directly instead of local storage compatibility bridges: $(paste -sd '; ' "$STANDALONE_THIN_COMPAT_BRIDGE_HITS_FILE")"
fi
(
cd "$ROOT_DIR"
{
for file in \
crates/iam/src/storage_compat.rs \
crates/heal/src/heal/storage_compat.rs \
crates/scanner/src/storage_compat.rs; do
[[ -e "$file" ]] && printf '%s:1:external owner bridge file exists\n' "$file"
done
rg -n --with-filename 'storage_compat' \
crates/iam/src \
crates/heal/src \
crates/scanner/src \
-g '*.rs' || true
}
) >"$EXTERNAL_OWNER_COMPAT_BRIDGE_HITS_FILE"
if [[ -s "$EXTERNAL_OWNER_COMPAT_BRIDGE_HITS_FILE" ]]; then
report_failure "IAM/heal/scanner consumers must import owner APIs directly instead of local storage compatibility bridges: $(paste -sd '; ' "$EXTERNAL_OWNER_COMPAT_BRIDGE_HITS_FILE")"
fi
(
cd "$ROOT_DIR"
rg -n --with-filename 'crate::storage_compat' \
@@ -1357,8 +1379,10 @@ fi
(
cd "$ROOT_DIR"
rg -n --no-heading 'pub\(crate\)\s+use rustfs_ecstore::api::bucket::\{[^}]*\b(?:bucket_target_sys|lifecycle|metadata_sys|replication|versioning|versioning_sys)\b[^}]*\}\s*;' \
crates/scanner/src/storage_compat.rs || true
if [[ -f crates/scanner/src/storage_compat.rs ]]; then
rg -n --no-heading 'pub\(crate\)\s+use rustfs_ecstore::api::bucket::\{[^}]*\b(?:bucket_target_sys|lifecycle|metadata_sys|replication|versioning|versioning_sys)\b[^}]*\}\s*;' \
crates/scanner/src/storage_compat.rs || true
fi
) >"$SCANNER_BUCKET_STORAGE_COMPAT_MODULE_HITS_FILE"
if [[ -s "$SCANNER_BUCKET_STORAGE_COMPAT_MODULE_HITS_FILE" ]]; then