mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-09 22:59:59 +00:00
refactor: add object store resolver for standalone crates (#3437)
This commit is contained in:
@@ -171,6 +171,21 @@ pub fn new_object_layer_fn() -> Option<Arc<ECStore>> {
|
||||
GLOBAL_OBJECT_API.get().cloned()
|
||||
}
|
||||
|
||||
pub type ObjectStoreResolver = dyn Fn() -> Option<Arc<ECStore>> + Send + Sync + 'static;
|
||||
|
||||
static GLOBAL_OBJECT_STORE_RESOLVER: OnceLock<Arc<ObjectStoreResolver>> = OnceLock::new();
|
||||
|
||||
pub fn set_object_store_resolver(resolver: Arc<ObjectStoreResolver>) -> bool {
|
||||
GLOBAL_OBJECT_STORE_RESOLVER.set(resolver).is_ok()
|
||||
}
|
||||
|
||||
pub fn resolve_object_store_handle() -> Option<Arc<ECStore>> {
|
||||
GLOBAL_OBJECT_STORE_RESOLVER
|
||||
.get()
|
||||
.and_then(|resolver| resolver())
|
||||
.or_else(new_object_layer_fn)
|
||||
}
|
||||
|
||||
/// Set the global object layer
|
||||
///
|
||||
/// # Arguments
|
||||
|
||||
@@ -54,10 +54,10 @@ mod pools_test;
|
||||
mod store_test;
|
||||
pub mod tier;
|
||||
|
||||
pub use global::new_object_layer_fn;
|
||||
pub use global::set_global_endpoints;
|
||||
pub use global::update_erasure_type;
|
||||
pub use global::{get_global_lock_client, get_global_lock_clients, set_global_lock_client, set_global_lock_clients};
|
||||
pub use global::{new_object_layer_fn, resolve_object_store_handle, set_object_store_resolver};
|
||||
|
||||
pub use global::GLOBAL_Endpoints;
|
||||
pub use store_api::StorageAPI;
|
||||
|
||||
@@ -316,7 +316,7 @@ impl NotifyConfigManager {
|
||||
where
|
||||
F: FnMut(&mut Config) -> bool,
|
||||
{
|
||||
let Some(store) = rustfs_ecstore::global::new_object_layer_fn() else {
|
||||
let Some(store) = rustfs_ecstore::global::resolve_object_store_handle() else {
|
||||
return Err(NotificationError::StorageNotAvailable(
|
||||
"Failed to save target configuration: server storage not initialized".to_string(),
|
||||
));
|
||||
|
||||
@@ -34,8 +34,8 @@ use rustfs_ecstore::bucket::metadata_sys::get_quota_config;
|
||||
use rustfs_ecstore::bucket::replication::GLOBAL_REPLICATION_STATS;
|
||||
use rustfs_ecstore::data_usage::load_data_usage_from_backend;
|
||||
use rustfs_ecstore::global::get_global_bucket_monitor;
|
||||
use rustfs_ecstore::new_object_layer_fn;
|
||||
use rustfs_ecstore::pools::{get_total_usable_capacity, get_total_usable_capacity_free};
|
||||
use rustfs_ecstore::resolve_object_store_handle;
|
||||
use rustfs_ecstore::store_api::BucketOperations;
|
||||
use rustfs_iam::{get_global_iam_sys, oidc::oidc_plugin_authn_metrics_snapshot};
|
||||
use rustfs_io_metrics::internode_metrics::global_internode_metrics;
|
||||
@@ -152,7 +152,7 @@ pub struct ProcessMetricBundle {
|
||||
|
||||
/// Collect cluster and cluster-health statistics from a single storage snapshot.
|
||||
pub async fn collect_cluster_and_health_stats() -> (ClusterStats, ClusterHealthStats) {
|
||||
let Some(store) = new_object_layer_fn() else {
|
||||
let Some(store) = resolve_object_store_handle() else {
|
||||
return (ClusterStats::default(), ClusterHealthStats::default());
|
||||
};
|
||||
|
||||
@@ -242,7 +242,7 @@ pub async fn collect_cluster_health_stats() -> ClusterHealthStats {
|
||||
|
||||
/// Collect bucket statistics from the storage layer.
|
||||
pub async fn collect_bucket_stats() -> Vec<BucketStats> {
|
||||
let Some(store) = new_object_layer_fn() else {
|
||||
let Some(store) = resolve_object_store_handle() else {
|
||||
return Vec::new();
|
||||
};
|
||||
|
||||
@@ -523,7 +523,7 @@ pub fn collect_system_memory_stats() -> MemoryStats {
|
||||
|
||||
/// Collect node disk stats and drive stats from a single storage snapshot.
|
||||
pub async fn collect_disk_and_system_drive_stats() -> (Vec<DiskStats>, Vec<DriveDetailedStats>, DriveCountStats) {
|
||||
let Some(store) = new_object_layer_fn() else {
|
||||
let Some(store) = resolve_object_store_handle() else {
|
||||
return (Vec::new(), Vec::new(), DriveCountStats::default());
|
||||
};
|
||||
|
||||
@@ -711,7 +711,7 @@ pub fn collect_internode_network_stats() -> Option<NetworkStats> {
|
||||
|
||||
/// Collect cluster config metrics from backend parity configuration.
|
||||
pub async fn collect_cluster_config_stats() -> Option<ClusterConfigStats> {
|
||||
let store = new_object_layer_fn()?;
|
||||
let store = resolve_object_store_handle()?;
|
||||
let backend = StorageAdminApi::backend_info(store.as_ref()).await;
|
||||
|
||||
Some(ClusterConfigStats {
|
||||
@@ -722,7 +722,7 @@ pub async fn collect_cluster_config_stats() -> Option<ClusterConfigStats> {
|
||||
|
||||
/// Collect cluster erasure set metrics from storage and backend topology info.
|
||||
pub async fn collect_erasure_set_stats() -> Vec<ErasureSetStats> {
|
||||
let Some(store) = new_object_layer_fn() else {
|
||||
let Some(store) = resolve_object_store_handle() else {
|
||||
return Vec::new();
|
||||
};
|
||||
|
||||
@@ -801,7 +801,7 @@ pub async fn collect_iam_stats() -> Option<IamStats> {
|
||||
/// builds cluster totals plus per-bucket distributions from the returned
|
||||
/// histograms. It does not trigger an inline object-data rescan.
|
||||
pub async fn collect_cluster_usage_metric_stats() -> Option<(ClusterUsageStats, Vec<BucketUsageStats>)> {
|
||||
let store = new_object_layer_fn()?;
|
||||
let store = resolve_object_store_handle()?;
|
||||
let data_usage = load_data_usage_from_backend(store.clone()).await.ok()?;
|
||||
let mut buckets = Vec::with_capacity(data_usage.buckets_usage.len());
|
||||
|
||||
|
||||
@@ -16,7 +16,7 @@
|
||||
|
||||
use super::{SwiftError, SwiftResult};
|
||||
use rustfs_credentials::Credentials;
|
||||
use rustfs_ecstore::new_object_layer_fn;
|
||||
use rustfs_ecstore::resolve_object_store_handle;
|
||||
use rustfs_ecstore::store_api::BucketOperations;
|
||||
use rustfs_storage_api::MakeBucketOptions;
|
||||
use s3s::dto::{Tag, Tagging};
|
||||
@@ -160,7 +160,7 @@ pub async fn update_account_metadata(
|
||||
) -> SwiftResult<()> {
|
||||
let bucket_name = get_account_metadata_bucket_name(account);
|
||||
|
||||
let Some(store) = new_object_layer_fn() else {
|
||||
let Some(store) = resolve_object_store_handle() else {
|
||||
return Err(SwiftError::InternalServerError("Storage layer not initialized".to_string()));
|
||||
};
|
||||
|
||||
|
||||
@@ -20,7 +20,7 @@ use super::account::validate_account_access;
|
||||
use super::types::Container;
|
||||
use super::{SwiftError, SwiftResult};
|
||||
use rustfs_credentials::Credentials;
|
||||
use rustfs_ecstore::new_object_layer_fn;
|
||||
use rustfs_ecstore::resolve_object_store_handle;
|
||||
use rustfs_ecstore::store_api::{BucketOperations, ListOperations};
|
||||
use rustfs_storage_api::{BucketInfo, BucketOptions, DeleteBucketOptions, MakeBucketOptions};
|
||||
use s3s::dto::{Tag, Tagging};
|
||||
@@ -242,7 +242,7 @@ pub async fn list_containers(account: &str, credentials: &Credentials) -> SwiftR
|
||||
let mapper = ContainerMapper::default();
|
||||
|
||||
// Get storage layer
|
||||
let Some(store) = new_object_layer_fn() else {
|
||||
let Some(store) = resolve_object_store_handle() else {
|
||||
return Err(SwiftError::InternalServerError("Storage layer not initialized".to_string()));
|
||||
};
|
||||
|
||||
@@ -300,7 +300,7 @@ pub async fn create_container(account: &str, container: &str, credentials: &Cred
|
||||
let bucket_name = mapper.swift_to_s3_bucket(container, &project_id);
|
||||
|
||||
// Get storage layer
|
||||
let Some(store) = new_object_layer_fn() else {
|
||||
let Some(store) = resolve_object_store_handle() else {
|
||||
return Err(SwiftError::InternalServerError("Storage layer not initialized".to_string()));
|
||||
};
|
||||
|
||||
@@ -394,7 +394,7 @@ pub async fn get_container_metadata(account: &str, container: &str, credentials:
|
||||
let bucket_name = mapper.swift_to_s3_bucket(container, &project_id);
|
||||
|
||||
// Get storage layer
|
||||
let Some(store) = new_object_layer_fn() else {
|
||||
let Some(store) = resolve_object_store_handle() else {
|
||||
return Err(SwiftError::InternalServerError("Storage layer not initialized".to_string()));
|
||||
};
|
||||
|
||||
@@ -471,7 +471,7 @@ pub async fn update_container_metadata(
|
||||
let bucket_name = mapper.swift_to_s3_bucket(container, &project_id);
|
||||
|
||||
// Get storage layer
|
||||
let Some(store) = new_object_layer_fn() else {
|
||||
let Some(store) = resolve_object_store_handle() else {
|
||||
return Err(SwiftError::InternalServerError("Storage layer not initialized".to_string()));
|
||||
};
|
||||
|
||||
@@ -570,7 +570,7 @@ pub async fn delete_container(account: &str, container: &str, credentials: &Cred
|
||||
let bucket_name = mapper.swift_to_s3_bucket(container, &project_id);
|
||||
|
||||
// Get storage layer
|
||||
let Some(store) = new_object_layer_fn() else {
|
||||
let Some(store) = resolve_object_store_handle() else {
|
||||
return Err(SwiftError::InternalServerError("Storage layer not initialized".to_string()));
|
||||
};
|
||||
|
||||
@@ -658,7 +658,7 @@ pub async fn list_objects(
|
||||
let bucket = mapper.swift_to_s3_bucket(container, &project_id);
|
||||
|
||||
// Get storage layer
|
||||
let Some(store) = new_object_layer_fn() else {
|
||||
let Some(store) = resolve_object_store_handle() else {
|
||||
return Err(SwiftError::InternalServerError("Storage layer not initialized".to_string()));
|
||||
};
|
||||
|
||||
@@ -768,7 +768,7 @@ pub async fn enable_versioning(
|
||||
let archive_bucket_name = mapper.swift_to_s3_bucket(archive_container, &project_id);
|
||||
|
||||
// Get storage layer
|
||||
let Some(store) = new_object_layer_fn() else {
|
||||
let Some(store) = resolve_object_store_handle() else {
|
||||
return Err(SwiftError::InternalServerError("Storage layer not initialized".to_string()));
|
||||
};
|
||||
|
||||
@@ -864,7 +864,7 @@ pub async fn disable_versioning(account: &str, container: &str, credentials: &Cr
|
||||
let bucket_name = mapper.swift_to_s3_bucket(container, &project_id);
|
||||
|
||||
// Verify container exists
|
||||
let Some(_store) = new_object_layer_fn() else {
|
||||
let Some(_store) = resolve_object_store_handle() else {
|
||||
return Err(SwiftError::InternalServerError("Storage layer not initialized".to_string()));
|
||||
};
|
||||
|
||||
@@ -1015,7 +1015,7 @@ pub async fn set_container_acl(
|
||||
let bucket_name = mapper.swift_to_s3_bucket(container, &project_id);
|
||||
|
||||
// Get storage layer
|
||||
let Some(store) = new_object_layer_fn() else {
|
||||
let Some(store) = resolve_object_store_handle() else {
|
||||
return Err(SwiftError::InternalServerError("Storage layer not initialized".to_string()));
|
||||
};
|
||||
|
||||
|
||||
@@ -54,7 +54,7 @@ use super::container::ContainerMapper;
|
||||
use super::{SwiftError, SwiftResult};
|
||||
use axum::http::HeaderMap;
|
||||
use rustfs_credentials::Credentials;
|
||||
use rustfs_ecstore::new_object_layer_fn;
|
||||
use rustfs_ecstore::resolve_object_store_handle;
|
||||
use rustfs_ecstore::store_api::{BucketOperations, ObjectIO, ObjectOperations, ObjectOptions, PutObjReader};
|
||||
use rustfs_rio::HashReader;
|
||||
use rustfs_storage_api::BucketOptions;
|
||||
@@ -375,7 +375,7 @@ where
|
||||
}
|
||||
|
||||
// 12. Get storage layer
|
||||
let Some(store) = new_object_layer_fn() else {
|
||||
let Some(store) = resolve_object_store_handle() else {
|
||||
return Err(SwiftError::InternalServerError("Storage layer not initialized".to_string()));
|
||||
};
|
||||
|
||||
@@ -455,7 +455,7 @@ where
|
||||
validate_metadata(metadata)?;
|
||||
|
||||
// Get storage layer
|
||||
let Some(store) = new_object_layer_fn() else {
|
||||
let Some(store) = resolve_object_store_handle() else {
|
||||
return Err(SwiftError::InternalServerError("Storage layer not initialized".to_string()));
|
||||
};
|
||||
|
||||
@@ -543,7 +543,7 @@ pub async fn get_object(
|
||||
let bucket = mapper.swift_to_s3_bucket(container, &project_id);
|
||||
|
||||
// 5. Get storage layer
|
||||
let Some(store) = new_object_layer_fn() else {
|
||||
let Some(store) = resolve_object_store_handle() else {
|
||||
return Err(SwiftError::InternalServerError("Storage layer not initialized".to_string()));
|
||||
};
|
||||
|
||||
@@ -603,7 +603,7 @@ pub async fn head_object(
|
||||
let bucket = mapper.swift_to_s3_bucket(container, &project_id);
|
||||
|
||||
// 5. Get storage layer
|
||||
let Some(store) = new_object_layer_fn() else {
|
||||
let Some(store) = resolve_object_store_handle() else {
|
||||
return Err(SwiftError::InternalServerError("Storage layer not initialized".to_string()));
|
||||
};
|
||||
|
||||
@@ -661,7 +661,7 @@ pub async fn delete_object(account: &str, container: &str, object: &str, credent
|
||||
let bucket = mapper.swift_to_s3_bucket(container, &project_id);
|
||||
|
||||
// 5. Get storage layer
|
||||
let Some(store) = new_object_layer_fn() else {
|
||||
let Some(store) = resolve_object_store_handle() else {
|
||||
return Err(SwiftError::InternalServerError("Storage layer not initialized".to_string()));
|
||||
};
|
||||
|
||||
@@ -724,7 +724,7 @@ pub async fn update_object_metadata(
|
||||
let bucket = mapper.swift_to_s3_bucket(container, &project_id);
|
||||
|
||||
// 5. Get storage layer
|
||||
let Some(store) = new_object_layer_fn() else {
|
||||
let Some(store) = resolve_object_store_handle() else {
|
||||
return Err(SwiftError::InternalServerError("Storage layer not initialized".to_string()));
|
||||
};
|
||||
|
||||
@@ -848,7 +848,7 @@ pub async fn copy_object(
|
||||
let dst_bucket = mapper.swift_to_s3_bucket(dst_container, &dst_project_id);
|
||||
|
||||
// 6. Get storage layer
|
||||
let Some(store) = new_object_layer_fn() else {
|
||||
let Some(store) = resolve_object_store_handle() else {
|
||||
return Err(SwiftError::InternalServerError("Storage layer not initialized".to_string()));
|
||||
};
|
||||
|
||||
|
||||
@@ -56,7 +56,7 @@ use super::container::ContainerMapper;
|
||||
use super::object::{ObjectKeyMapper, head_object};
|
||||
use super::{SwiftError, SwiftResult};
|
||||
use rustfs_credentials::Credentials;
|
||||
use rustfs_ecstore::new_object_layer_fn;
|
||||
use rustfs_ecstore::resolve_object_store_handle;
|
||||
use rustfs_ecstore::store_api::{ListOperations, ObjectOperations, ObjectOptions};
|
||||
use std::time::{SystemTime, UNIX_EPOCH};
|
||||
use tracing::{debug, error};
|
||||
@@ -197,7 +197,7 @@ pub async fn archive_current_version(
|
||||
let version_key = ObjectKeyMapper::swift_to_s3_key(&version_name)?;
|
||||
|
||||
// Get storage layer
|
||||
let Some(store) = new_object_layer_fn() else {
|
||||
let Some(store) = resolve_object_store_handle() else {
|
||||
return Err(SwiftError::InternalServerError("Storage layer not initialized".to_string()));
|
||||
};
|
||||
|
||||
@@ -340,7 +340,7 @@ pub async fn restore_previous_version(
|
||||
let version_key = ObjectKeyMapper::swift_to_s3_key(newest_version)?;
|
||||
|
||||
// Get storage layer
|
||||
let Some(store) = new_object_layer_fn() else {
|
||||
let Some(store) = resolve_object_store_handle() else {
|
||||
return Err(SwiftError::InternalServerError("Storage layer not initialized".to_string()));
|
||||
};
|
||||
|
||||
@@ -476,7 +476,7 @@ pub async fn list_object_versions(
|
||||
let archive_bucket = mapper.swift_to_s3_bucket(archive_container, &project_id);
|
||||
|
||||
// Get storage layer
|
||||
let Some(store) = new_object_layer_fn() else {
|
||||
let Some(store) = resolve_object_store_handle() else {
|
||||
return Err(SwiftError::InternalServerError("Storage layer not initialized".to_string()));
|
||||
};
|
||||
|
||||
|
||||
@@ -26,7 +26,7 @@ use object_store::{
|
||||
use pin_project_lite::pin_project;
|
||||
use rustfs_common::DEFAULT_DELIMITER;
|
||||
use rustfs_ecstore::error::{StorageError, is_err_bucket_not_found, is_err_object_not_found, is_err_version_not_found};
|
||||
use rustfs_ecstore::new_object_layer_fn;
|
||||
use rustfs_ecstore::resolve_object_store_handle;
|
||||
use rustfs_ecstore::set_disk::DEFAULT_READ_BUFFER_SIZE;
|
||||
use rustfs_ecstore::store::ECStore;
|
||||
use rustfs_ecstore::store_api::{GetObjectReader, HTTPRangeSpec, ObjectIO, ObjectOperations, ObjectOptions};
|
||||
@@ -107,7 +107,7 @@ pub struct InvalidScanRange;
|
||||
|
||||
impl EcObjectStore {
|
||||
pub fn new(input: Arc<SelectObjectContentInput>) -> S3Result<Self> {
|
||||
let Some(store) = new_object_layer_fn() else {
|
||||
let Some(store) = resolve_object_store_handle() else {
|
||||
return Err(s3_error!(InternalError, "ec store not inited"));
|
||||
};
|
||||
|
||||
|
||||
@@ -39,7 +39,7 @@ use rustfs_ecstore::disk::error::DiskError;
|
||||
use rustfs_ecstore::disk::{Disk, DiskAPI};
|
||||
use rustfs_ecstore::error::{Error, StorageError};
|
||||
use rustfs_ecstore::global::GLOBAL_TierConfigMgr;
|
||||
use rustfs_ecstore::new_object_layer_fn;
|
||||
use rustfs_ecstore::resolve_object_store_handle;
|
||||
use rustfs_ecstore::set_disk::SetDisks;
|
||||
use rustfs_ecstore::store_api::{BucketOperations, ObjectIO, ObjectInfo};
|
||||
use rustfs_ecstore::{error::Result, store::ECStore};
|
||||
@@ -1333,7 +1333,7 @@ impl ScannerIODisk for Disk {
|
||||
|
||||
// TODO: object lock
|
||||
|
||||
let Some(ecstore) = new_object_layer_fn() else {
|
||||
let Some(ecstore) = resolve_object_store_handle() else {
|
||||
error!(
|
||||
target: "rustfs::scanner::io",
|
||||
event = EVENT_SCANNER_DISK_BUCKET_STATE,
|
||||
|
||||
Reference in New Issue
Block a user