gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-07 22:03:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(runtime): route RustFS runtime consumers through storage owner (#3756)

Browse Source
This commit is contained in:
Zhengchao An
2026-06-23 05:17:56 +08:00
committed by GitHub
parent e0b79aa00c
commit 198fd4f150
50 changed files with 1622 additions and 384 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/admin/mod.rs
+72 -15
View File
@@ -103,21 +103,78 @@ fn register_admin_routes(r: &mut S3Router<AdminOperation>) -> std::io::Result<()
use std::ops::Deref;
use std::sync::Arc;
use rustfs_ecstore::api::bucket as ecstore_bucket;
use rustfs_ecstore::api::capacity as ecstore_capacity;
use rustfs_ecstore::api::client as ecstore_client;
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::layout as ecstore_layout;
use rustfs_ecstore::api::metrics as ecstore_metrics;
use rustfs_ecstore::api::notification as ecstore_notification;
use rustfs_ecstore::api::rebalance as ecstore_rebalance;
use rustfs_ecstore::api::rpc as ecstore_rpc;
use rustfs_ecstore::api::storage as ecstore_storage;
use rustfs_ecstore::api::tier as ecstore_tier;
mod ecstore_bucket {
pub(crate) use crate::storage::ecstore_bucket::{
bandwidth, bucket_target_sys, lifecycle, metadata, metadata_sys, quota, replication, target, utils, versioning,
versioning_sys,
};
}
mod ecstore_capacity {
pub(crate) use crate::storage::ecstore_capacity::is_reserved_or_invalid_bucket;
}
mod ecstore_client {
pub(crate) use crate::storage::ecstore_client::admin_handler_utils;
}
mod ecstore_config {
pub(crate) use crate::storage::ecstore_config::{com, init, set_global_storage_class, storageclass};
}
mod ecstore_data_usage {
pub(crate) use crate::storage::ecstore_data_usage::load_data_usage_from_backend;
}
#[allow(unused_imports)]
mod ecstore_disk {
pub(crate) use crate::storage::ecstore_disk::{RUSTFS_META_BUCKET, endpoint};
}
mod ecstore_error {
pub(crate) use crate::storage::ecstore_error::StorageError;
}
mod ecstore_global {
pub(crate) use crate::storage::ecstore_global::{
GLOBAL_BOOT_TIME, get_global_bucket_monitor, get_global_deployment_id, get_global_endpoints_opt, get_global_region,
global_rustfs_port,
};
}
#[allow(unused_imports)]
mod ecstore_layout {
pub(crate) use crate::storage::ecstore_layout::{EndpointServerPools, Endpoints, PoolEndpoints};
}
mod ecstore_metrics {
pub(crate) use crate::storage::ecstore_metrics::{CollectMetricsOpts, MetricType, collect_local_metrics};
}
mod ecstore_notification {
pub(crate) use crate::storage::ecstore_notification::{NotificationSys, get_global_notification_sys};
}
#[allow(unused_imports)]
mod ecstore_rebalance {
pub(crate) use crate::storage::ecstore_rebalance::{
DiskStat, RebalSaveOpt, RebalStatus, RebalanceCleanupWarningEntry, RebalanceCleanupWarnings, RebalanceInfo,
RebalanceMeta, RebalanceStats, RebalanceStopPropagationRecord, decode_rebalance_stop_propagation_record,
encode_rebalance_stop_propagation_record,
};
}
mod ecstore_rpc {
pub(crate) use crate::storage::ecstore_rpc::PeerRestClient;
}
mod ecstore_storage {
pub(crate) use crate::storage::ecstore_storage::ECStore;
}
mod ecstore_tier {
pub(crate) use crate::storage::ecstore_tier::{tier, tier_admin, tier_config, tier_handlers};
}
pub(crate) const RUSTFS_META_BUCKET: &str = ecstore_disk::RUSTFS_META_BUCKET;
pub(crate) const STORAGE_CLASS_SUB_SYS: &str = ecstore_config::com::STORAGE_CLASS_SUB_SYS;
rustfs/src/app/mod.rs
+83 -53
View File
@@ -43,54 +43,84 @@ mod lifecycle_transition_api_test;
use std::sync::Arc;
use rustfs_ecstore::api::admin as ecstore_admin;
use rustfs_ecstore::api::bucket as ecstore_bucket;
use rustfs_ecstore::api::capacity as ecstore_capacity;
use rustfs_ecstore::api::client as ecstore_client;
use rustfs_ecstore::api::compression as ecstore_compression;
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::layout as ecstore_layout;
use rustfs_ecstore::api::notification as ecstore_notification;
use rustfs_ecstore::api::rio as ecstore_rio;
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;
mod ecstore_admin {
pub(crate) use crate::storage::ecstore_admin::get_server_info;
}
mod ecstore_bucket {
pub(crate) use crate::storage::ecstore_bucket::{
bucket_target_sys, lifecycle, metadata, metadata_sys, object_lock, policy_sys, quota, replication, tagging, target,
utils, versioning, versioning_sys,
};
}
mod ecstore_capacity {
pub(crate) use crate::storage::ecstore_capacity::{
PoolDecommissionInfo, PoolStatus, get_total_usable_capacity, get_total_usable_capacity_free,
};
}
#[allow(unused_imports)]
mod ecstore_client {
pub(crate) use crate::storage::ecstore_client::{object_api_utils, transition_api};
}
mod ecstore_compression {
pub(crate) use crate::storage::ecstore_compression::{MIN_DISK_COMPRESSIBLE_SIZE, is_disk_compressible};
}
mod ecstore_config {
pub(crate) use crate::storage::ecstore_config::storageclass;
}
mod ecstore_data_usage {
pub(crate) use crate::storage::ecstore_data_usage::{
apply_bucket_usage_memory_overlay, load_data_usage_from_backend, record_bucket_object_delete_memory,
record_bucket_object_write_memory,
};
}
#[cfg(test)]
mod ecstore_global {
pub(crate) use crate::storage::ecstore_global::GLOBAL_TierConfigMgr;
}
#[allow(unused_imports)]
mod ecstore_tier {
pub(crate) use crate::storage::ecstore_tier::{tier, tier_config, warm_backend};
}
pub(crate) const MIN_DISK_COMPRESSIBLE_SIZE: usize = ecstore_compression::MIN_DISK_COMPRESSIBLE_SIZE;
pub(crate) type DiskError = ecstore_disk::error::DiskError;
pub(crate) type DynReader = ecstore_rio::DynReader;
pub(crate) type ECStore = ecstore_storage::ECStore;
pub(crate) type EndpointServerPools = ecstore_layout::EndpointServerPools;
pub(crate) type HashReader = ecstore_rio::HashReader;
pub(crate) type NotificationSys = ecstore_notification::NotificationSys;
pub(crate) type ObjectStoreResolver = dyn Fn() -> Option<Arc<ECStore>> + Send + Sync + 'static;
pub(crate) type DiskError = crate::storage::DiskError;
pub(crate) type DynReader = crate::storage::DynReader;
pub(crate) type ECStore = crate::storage::ECStore;
pub(crate) type EndpointServerPools = crate::storage::EndpointServerPools;
pub(crate) type HashReader = crate::storage::HashReader;
pub(crate) type NotificationSys = crate::storage::NotificationSys;
pub(crate) type ObjectStoreResolver = crate::storage::ObjectStoreResolver;
pub(crate) type ObjectInfo = <ECStore as rustfs_storage_api::ObjectOperations>::ObjectInfo;
pub(crate) type ObjectOptions = <ECStore as rustfs_storage_api::ObjectOperations>::ObjectOptions;
pub(crate) type PoolDecommissionInfo = ecstore_capacity::PoolDecommissionInfo;
pub(crate) type PoolStatus = ecstore_capacity::PoolStatus;
pub(crate) type StorageError = ecstore_error::StorageError;
pub(crate) type StorageError = crate::storage::StorageError;
pub(crate) type Error = StorageError;
pub(crate) type TierConfigMgr = ecstore_tier::tier::TierConfigMgr;
pub(crate) type WriteEncryption = ecstore_rio::WriteEncryption;
pub(crate) type WritePlan = ecstore_rio::WritePlan;
pub(crate) type TierConfigMgr = crate::storage::TierConfigMgr;
pub(crate) type WriteEncryption = crate::storage::WriteEncryption;
pub(crate) type WritePlan = crate::storage::WritePlan;
#[cfg(test)]
pub(crate) type DecryptReader<R> = ecstore_rio::DecryptReader<R>;
pub(crate) type DecryptReader<R> = crate::storage::DecryptReader<R>;
#[cfg(test)]
pub(crate) type EncryptReader<R> = ecstore_rio::EncryptReader<R>;
pub(crate) type EncryptReader<R> = crate::storage::EncryptReader<R>;
#[cfg(test)]
pub(crate) type Endpoint = ecstore_disk::endpoint::Endpoint;
pub(crate) type Endpoint = crate::storage::Endpoint;
#[cfg(test)]
pub(crate) type Endpoints = ecstore_layout::Endpoints;
pub(crate) type Endpoints = crate::storage::Endpoints;
#[cfg(test)]
pub(crate) type HardLimitReader<R> = ecstore_rio::HardLimitReader<R>;
pub(crate) type HardLimitReader<R> = crate::storage::HardLimitReader<R>;
#[cfg(test)]
pub(crate) type PoolEndpoints = ecstore_layout::PoolEndpoints;
pub(crate) type PoolEndpoints = crate::storage::PoolEndpoints;
#[cfg(test)]
pub(crate) type TierConfig = ecstore_tier::tier_config::TierConfig;
#[cfg(test)]
@@ -103,7 +133,7 @@ pub(crate) type WarmBackendGetOpts = ecstore_tier::warm_backend::WarmBackendGetO
#[cfg(test)]
#[allow(non_snake_case)]
pub(crate) fn EndpointServerPools(pools: Vec<PoolEndpoints>) -> EndpointServerPools {
ecstore_layout::EndpointServerPools::from(pools)
crate::storage::EndpointServerPools::from(pools)
}
pub(crate) trait AppObjectLockConfigExt {
@@ -396,7 +426,7 @@ pub(crate) mod metadata_sys {
pub(crate) mod object_api_utils {
pub(crate) fn to_s3s_etag(etag: &str) -> s3s::dto::ETag {
super::ecstore_client::object_api_utils::to_s3s_etag(etag)
crate::storage::to_s3s_etag(etag)
}
}
@@ -565,19 +595,19 @@ pub(crate) async fn record_bucket_object_write_memory(bucket: &str, previous_cur
}
pub(crate) fn is_all_buckets_not_found(errs: &[Option<DiskError>]) -> bool {
ecstore_disk::error_reduce::is_all_buckets_not_found(errs)
crate::storage::is_all_buckets_not_found(errs)
}
pub(crate) fn is_err_bucket_not_found(err: &Error) -> bool {
ecstore_error::is_err_bucket_not_found(err)
crate::storage::is_err_bucket_not_found(err)
}
pub(crate) fn is_err_object_not_found(err: &Error) -> bool {
ecstore_error::is_err_object_not_found(err)
crate::storage::is_err_object_not_found(err)
}
pub(crate) fn is_err_version_not_found(err: &Error) -> bool {
ecstore_error::is_err_version_not_found(err)
crate::storage::is_err_version_not_found(err)
}
#[cfg(test)]
@@ -597,57 +627,57 @@ impl std::ops::Deref for GlobalTierConfigMgrCompat {
}
pub(crate) fn get_global_endpoints_opt() -> Option<EndpointServerPools> {
ecstore_global::get_global_endpoints_opt()
crate::storage::get_global_endpoints_opt()
}
pub(crate) fn get_global_region() -> Option<s3s::region::Region> {
ecstore_global::get_global_region()
crate::storage::get_global_region()
}
pub(crate) fn get_global_tier_config_mgr() -> Arc<tokio::sync::RwLock<TierConfigMgr>> {
ecstore_global::get_global_tier_config_mgr()
crate::storage::get_global_tier_config_mgr()
}
pub(crate) fn new_object_layer_fn() -> Option<Arc<ECStore>> {
ecstore_global::new_object_layer_fn()
crate::storage::new_object_layer_fn()
}
pub(crate) fn set_object_store_resolver(resolver: Arc<ObjectStoreResolver>) -> bool {
ecstore_global::set_object_store_resolver(resolver)
crate::storage::set_object_store_resolver(resolver)
}
pub(crate) fn get_global_notification_sys() -> Option<&'static NotificationSys> {
ecstore_notification::get_global_notification_sys()
crate::storage::get_global_notification_sys()
}
#[cfg(test)]
pub(crate) fn boxed_reader<R>(reader: R) -> DynReader
where
R: ecstore_rio::Reader + 'static,
R: crate::storage::ecstore_rio::Reader + 'static,
{
ecstore_rio::boxed_reader(reader)
crate::storage::boxed_reader(reader)
}
pub(crate) fn compression_metadata_value(algorithm: rustfs_utils::CompressionAlgorithm) -> String {
ecstore_rio::compression_metadata_value(algorithm)
crate::storage::compression_metadata_value(algorithm)
}
pub(crate) fn wrap_reader<R>(reader: R) -> DynReader
where
R: ecstore_rio::ReadStream + 'static,
R: crate::storage::ecstore_rio::ReadStream + 'static,
{
ecstore_rio::wrap_reader(reader)
crate::storage::wrap_reader(reader)
}
pub(crate) fn get_lock_acquire_timeout() -> tokio::time::Duration {
ecstore_set_disk::get_lock_acquire_timeout()
crate::storage::get_lock_acquire_timeout()
}
pub(crate) fn is_valid_storage_class(storage_class: &str) -> bool {
ecstore_set_disk::is_valid_storage_class(storage_class)
crate::storage::is_valid_storage_class(storage_class)
}
#[cfg(test)]
pub(crate) async fn init_local_disks(endpoint_pools: EndpointServerPools) -> Result<(), Error> {
ecstore_storage::init_local_disks(endpoint_pools).await
crate::storage::init_local_disks(endpoint_pools).await
}
rustfs/src/capacity/service.rs
+3 -8
View File
@@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use crate::storage::{ecstore_disk, ecstore_storage};
use crate::storage::{all_local_disk, disk_drive_path, disk_endpoint};
use rustfs_io_metrics::capacity_metrics::{
record_capacity_cache_hit, record_capacity_cache_miss, record_capacity_cache_served, record_capacity_refresh_request,
record_capacity_scan_mode,
@@ -263,7 +263,7 @@ pub async fn init_capacity_management_for_local_disks() {
"Capacity manager state changed"
);
let disks = ecstore_storage::all_local_disk().await;
let disks = all_local_disk().await;
if disks.is_empty() {
warn!(
component = LOG_COMPONENT_CAPACITY,
@@ -286,12 +286,7 @@ pub async fn init_capacity_management_for_local_disks() {
let disk_refs = disks
.iter()
.map(|ds| {
capacity_disk_ref(
ecstore_disk::DiskAPI::endpoint(ds.as_ref()).to_string(),
ecstore_disk::DiskAPI::to_string(ds.as_ref()),
)
})
.map(|ds| capacity_disk_ref(disk_endpoint(ds), disk_drive_path(ds)))
.collect();
info!(
rustfs/src/config/config_test.rs
+2 -2
View File
@@ -16,7 +16,7 @@
#[allow(unsafe_op_in_unsafe_fn)]
mod tests {
use crate::config::{CommandResult, Config, Opt, TlsCommands};
use crate::storage::ecstore_layout::DisksLayout;
use crate::storage::DisksLayout;
use rustfs_config::{DEFAULT_CONSOLE_ADDRESS, DEFAULT_CONSOLE_ENABLE, DEFAULT_OBS_ENDPOINT, RUSTFS_REGION};
use rustfs_credentials::{DEFAULT_ACCESS_KEY, DEFAULT_SECRET_KEY};
use serial_test::serial;
@@ -262,7 +262,7 @@ mod tests {
#[test]
#[serial]
fn test_volumes_and_disk_layout_parsing() {
use crate::storage::ecstore_layout::DisksLayout;
use crate::storage::DisksLayout;
// Test case 1: Single volume path
let args = vec!["rustfs", "/data/vol1"];
rustfs/src/error.rs
+1 -1
View File
@@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use crate::storage::{ecstore_bucket::quota::QuotaError, ecstore_error::StorageError};
use crate::storage::{QuotaError, StorageError};
use rustfs_storage_api::HTTPRangeError;
use s3s::{S3Error, S3ErrorCode};
rustfs/src/init.rs
+17 -17
View File
@@ -13,8 +13,10 @@
// limitations under the License.
use crate::server::ShutdownHandle;
use crate::storage::{ecstore_bucket::metadata_sys as ecstore_metadata_sys, ecstore_global};
use crate::storage::{process_lambda_configurations, process_queue_configurations, process_topic_configurations};
use crate::storage::{
get_bucket_notification_config, get_global_region, process_lambda_configurations, process_queue_configurations,
process_topic_configurations,
};
use crate::{admin, config, version};
use rustfs_config::{
DEFAULT_BUFFER_MAX_SIZE, DEFAULT_BUFFER_MIN_SIZE, DEFAULT_BUFFER_PROFILE, DEFAULT_BUFFER_UNKNOWN_SIZE, DEFAULT_UPDATE_CHECK,
@@ -157,7 +159,7 @@ fn arn_to_target_id(arn_str: &str) -> Result<rustfs_targets::arn::TargetID, Targ
/// * `buckets` - A vector of bucket names to process
#[instrument(skip_all)]
pub async fn add_bucket_notification_configuration(buckets: Vec<String>) {
let global_region = ecstore_global::get_global_region();
let global_region = get_global_region();
let region = global_region
.as_ref()
.filter(|r| !r.as_str().is_empty())
@@ -174,20 +176,18 @@ pub async fn add_bucket_notification_configuration(buckets: Vec<String>) {
RUSTFS_REGION
});
for bucket in buckets.iter() {
let has_notification_config = ecstore_metadata_sys::get_notification_config(bucket)
.await
.unwrap_or_else(|err| {
warn!(
target: "rustfs::init",
event = "notification_config_load_failed",
component = LOG_COMPONENT_INIT,
subsystem = LOG_SUBSYSTEM_NOTIFICATION,
bucket = %bucket,
error = ?err,
"Failed to load bucket notification configuration"
);
None
});
let has_notification_config = get_bucket_notification_config(bucket).await.unwrap_or_else(|err| {
warn!(
target: "rustfs::init",
event = "notification_config_load_failed",
component = LOG_COMPONENT_INIT,
subsystem = LOG_SUBSYSTEM_NOTIFICATION,
bucket = %bucket,
error = ?err,
"Failed to load bucket notification configuration"
);
None
});
match has_notification_config {
Some(cfg) => {
rustfs/src/runtime_capabilities.rs
+3 -6
View File
@@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use crate::storage::{ecstore_cluster, ecstore_layout::EndpointServerPools};
use crate::storage::{EndpointServerPools, topology_snapshot_from_endpoint_pools_with_capabilities};
use rustfs_storage_api::{
CapabilitySnapshotError, CapabilityStatus, DiskCapabilities, MemorySamplingState, ObservabilitySnapshot,
ObservabilitySnapshotProvider, PlatformSupport, TopologyCapabilities, TopologySnapshot, TopologySnapshotProvider,
@@ -79,7 +79,7 @@ impl TopologySnapshotProvider for EndpointTopologySnapshotProvider {
}
pub fn topology_snapshot_from_endpoint_pools(endpoint_pools: &EndpointServerPools) -> TopologySnapshot {
ecstore_cluster::topology_snapshot_from_endpoint_pools_with_capabilities(
topology_snapshot_from_endpoint_pools_with_capabilities(
endpoint_pools,
TopologyCapabilities {
profiling: cpu_profiling_status(),
@@ -131,10 +131,7 @@ fn cgroup_memory_status() -> CapabilityStatus {
#[cfg(test)]
mod tests {
use super::*;
use crate::storage::{
ecstore_disk::endpoint::Endpoint,
ecstore_layout::{Endpoints, PoolEndpoints},
};
use crate::storage::{Endpoint, Endpoints, PoolEndpoints};
use rustfs_storage_api::{CapabilityState, ObservabilitySnapshotProvider, TopologySnapshotProvider};
#[tokio::test]
rustfs/src/server/event.rs
+1 -2
View File
@@ -14,8 +14,7 @@
use super::{module_switch::resolve_notify_module_state, refresh_persisted_module_switches_from_store};
use crate::app::context::resolve_server_config;
use crate::storage::StorageObjectInfo;
use crate::storage::ecstore_event::{EventArgs as EcstoreEventArgs, register_event_dispatch_hook};
use crate::storage::{EventArgs as EcstoreEventArgs, StorageObjectInfo, register_event_dispatch_hook};
use chrono::{DateTime, Utc};
use rustfs_notify::{EventArgs as NotifyEventArgs, NotifyObjectInfo};
use rustfs_s3_types::EventName;
rustfs/src/server/http.rs
+1 -1
View File
@@ -31,9 +31,9 @@ use crate::server::{
},
};
use crate::storage;
use crate::storage::ecstore_rpc::{TONIC_RPC_PREFIX, verify_rpc_signature};
use crate::storage::rpc::InternodeRpcService;
use crate::storage::tonic_service::make_server;
use crate::storage::{TONIC_RPC_PREFIX, verify_rpc_signature};
use bytes::Bytes;
use http::{HeaderMap, Method, Request as HttpRequest, Response};
use hyper_util::{
rustfs/src/server/module_switch.rs
+3 -3
View File
@@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use crate::storage::{ecstore_config, ecstore_error::Error as StorageError, ecstore_global::resolve_object_store_handle};
use crate::storage::{Error as StorageError, read_config, resolve_object_store_handle, save_config};
use serde::{Deserialize, Serialize};
use std::sync::atomic::{AtomicBool, Ordering};
@@ -162,7 +162,7 @@ pub(crate) async fn refresh_persisted_module_switches_from_store() -> Result<Per
return Err("storage layer not initialized".to_string());
};
let (config, configured) = match ecstore_config::com::read_config(store, MODULE_SWITCH_CONFIG_PATH).await {
let (config, configured) = match read_config(store, MODULE_SWITCH_CONFIG_PATH).await {
Ok(data) => (
serde_json::from_slice::<PersistedModuleSwitches>(&data)
.map_err(|e| format!("failed to deserialize module switch config: {e}"))?,
@@ -184,7 +184,7 @@ pub(crate) async fn save_persisted_module_switches_to_store(config: PersistedMod
};
let data = serde_json::to_vec(&config).map_err(|e| format!("failed to serialize module switch config: {e}"))?;
ecstore_config::com::save_config(store, MODULE_SWITCH_CONFIG_PATH, data)
save_config(store, MODULE_SWITCH_CONFIG_PATH, data)
.await
.map_err(|e| format!("failed to save module switch config: {e}"))?;
rustfs/src/server/readiness.rs
+4 -5
View File
@@ -14,13 +14,12 @@
use crate::server::{ServiceState, ServiceStateManager};
use crate::server::{has_path_prefix, is_table_catalog_path};
#[cfg(test)]
use crate::storage::ecstore_layout::{Endpoints, PoolEndpoints};
use crate::storage::{
ecstore_disk::endpoint::Endpoint,
ecstore_global::{get_global_endpoints_opt, get_global_lock_clients, is_dist_erasure, resolve_object_store_handle},
ecstore_layout::EndpointServerPools,
Endpoint, EndpointServerPools, get_global_endpoints_opt, get_global_lock_clients, is_dist_erasure,
resolve_object_store_handle,
};
#[cfg(test)]
use crate::storage::{Endpoints, PoolEndpoints};
use bytes::Bytes;
use http::{Request as HttpRequest, Response, StatusCode};
use http_body::Body;
rustfs/src/startup_bucket_metadata.rs
+8 -9
View File
@@ -13,8 +13,7 @@
// limitations under the License.
use crate::storage::{
ECStore,
ecstore_bucket::{metadata_sys as ecstore_metadata_sys, migration as ecstore_migration, replication as ecstore_replication},
ECStore, get_global_replication_pool, init_bucket_metadata_sys, try_migrate_bucket_metadata, try_migrate_iam_config,
};
use rustfs_storage_api::{BucketOperations, BucketOptions};
use std::{
@@ -34,9 +33,9 @@ pub(crate) async fn init_embedded_bucket_metadata_runtime(store: Arc<ECStore>) -
let buckets: Vec<String> = buckets_list.into_iter().map(|v| v.name).collect();
ecstore_migration::try_migrate_bucket_metadata(store.clone()).await;
ecstore_metadata_sys::init_bucket_metadata_sys(store.clone(), buckets.clone()).await;
ecstore_migration::try_migrate_iam_config(store).await;
try_migrate_bucket_metadata(store.clone()).await;
init_bucket_metadata_sys(store.clone(), buckets.clone()).await;
try_migrate_iam_config(store).await;
Ok(buckets)
}
@@ -52,14 +51,14 @@ pub(crate) async fn init_bucket_metadata_runtime(store: Arc<ECStore>, ctx: Cance
let buckets: Vec<String> = buckets_list.into_iter().map(|v| v.name).collect();
ecstore_migration::try_migrate_bucket_metadata(store.clone()).await;
try_migrate_bucket_metadata(store.clone()).await;
if let Some(pool) = ecstore_replication::get_global_replication_pool() {
if let Some(pool) = get_global_replication_pool() {
pool.init_resync(ctx, buckets.clone()).await?;
}
ecstore_migration::try_migrate_iam_config(store.clone()).await;
ecstore_metadata_sys::init_bucket_metadata_sys(store, buckets.clone()).await;
try_migrate_iam_config(store.clone()).await;
init_bucket_metadata_sys(store, buckets.clone()).await;
Ok(buckets)
}
rustfs/src/startup_fs_guard.rs
+1 -1
View File
@@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use crate::storage::ecstore_layout::EndpointServerPools;
use crate::storage::EndpointServerPools;
use rustfs_config::{
DEFAULT_RUSTFS_UNSUPPORTED_FS_POLICY, ENV_RUSTFS_UNSUPPORTED_FS_POLICY, RUSTFS_UNSUPPORTED_FS_POLICY_FAIL,
RUSTFS_UNSUPPORTED_FS_POLICY_WARN,
rustfs/src/startup_notification.rs
+6 -6
View File
@@ -13,7 +13,7 @@
// limitations under the License.
use crate::init::add_bucket_notification_configuration;
use crate::storage::{ecstore_error, ecstore_layout::EndpointServerPools, ecstore_notification};
use crate::storage::{EndpointServerPools, Result as StorageResult, new_global_notification_sys};
use std::{
future::Future,
io::{Error, Result},
@@ -50,14 +50,14 @@ pub(crate) async fn init_notification_runtime(endpoint_pools: EndpointServerPool
})
}
pub(crate) async fn init_notification_system(endpoint_pools: EndpointServerPools) -> ecstore_error::Result<()> {
init_notification_system_with(|| ecstore_notification::new_global_notification_sys(endpoint_pools)).await
pub(crate) async fn init_notification_system(endpoint_pools: EndpointServerPools) -> StorageResult<()> {
init_notification_system_with(|| new_global_notification_sys(endpoint_pools)).await
}
async fn init_notification_system_with<InitFn, InitFuture>(init_notification: InitFn) -> ecstore_error::Result<()>
async fn init_notification_system_with<InitFn, InitFuture>(init_notification: InitFn) -> StorageResult<()>
where
InitFn: FnOnce() -> InitFuture,
InitFuture: Future<Output = ecstore_error::Result<()>>,
InitFuture: Future<Output = StorageResult<()>>,
{
init_notification().await
}
@@ -76,7 +76,7 @@ fn log_embedded_optional_service_skipped(service: &str, err: impl std::fmt::Disp
#[cfg(test)]
mod tests {
use super::init_notification_system_with;
use crate::storage::ecstore_error::Error as EcstoreError;
use crate::storage::Error as EcstoreError;
#[tokio::test]
async fn notification_system_returns_source_error() {
rustfs/src/startup_server.rs
+1 -1
View File
@@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use crate::storage::ecstore_global::{set_global_region, set_global_rustfs_port};
use crate::storage::{set_global_region, set_global_rustfs_port};
use crate::{
capacity::capacity_integration::init_capacity_management,
config::Config,
rustfs/src/startup_services.rs
+1 -1
View File
@@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use crate::storage::{ECStore, ecstore_layout::EndpointServerPools};
use crate::storage::{ECStore, EndpointServerPools};
use crate::{
config::Config,
init::{init_buffer_profile_system, init_kms_system},
rustfs/src/startup_shutdown.rs
+1 -1
View File
@@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use crate::storage::ecstore_global::shutdown_background_services;
use crate::storage::shutdown_background_services;
use crate::{
server::{ServiceState, ServiceStateManager, ShutdownHandle, ShutdownSignal, shutdown_event_notifier, stop_audit_system},
startup_optional_runtime_sidecars::{
rustfs/src/startup_storage.rs
+19 -20
View File
@@ -13,10 +13,9 @@
// limitations under the License.
use crate::startup_fs_guard::enforce_unsupported_fs_policy;
use crate::storage::ECStore;
use crate::storage::{
ecstore_bucket::replication as ecstore_replication, ecstore_config, ecstore_global, ecstore_layout::EndpointServerPools,
ecstore_storage,
ECStore, EndpointServerPools, init_background_replication, init_ecstore_config, init_global_config_sys, init_local_disks,
init_lock_clients, prewarm_local_disk_id_map, set_global_endpoints, try_migrate_server_config, update_erasure_type,
};
use rustfs_common::{GlobalReadiness, SystemStage};
use std::{
@@ -71,8 +70,8 @@ pub(crate) async fn init_startup_storage_foundation(server_address: &str, volume
.map_err(Error::other)?;
enforce_unsupported_fs_policy(&endpoint_pools)?;
ecstore_global::set_global_endpoints(endpoint_pools.as_ref().clone());
ecstore_global::update_erasure_type(setup_type).await;
set_global_endpoints(endpoint_pools.as_ref().clone());
update_erasure_type(setup_type).await;
debug!(
target: "rustfs::main::run",
@@ -83,7 +82,7 @@ pub(crate) async fn init_startup_storage_foundation(server_address: &str, volume
state = "starting",
"starting local disk initialization"
);
ecstore_storage::init_local_disks(endpoint_pools.clone())
init_local_disks(endpoint_pools.clone())
.await
.inspect_err(|err| {
error!(
@@ -98,8 +97,8 @@ pub(crate) async fn init_startup_storage_foundation(server_address: &str, volume
);
})
.map_err(Error::other)?;
ecstore_storage::prewarm_local_disk_id_map().await;
ecstore_storage::init_lock_clients(endpoint_pools.clone());
prewarm_local_disk_id_map().await;
init_lock_clients(endpoint_pools.clone());
log_storage_pool_layout(&endpoint_pools);
@@ -115,13 +114,13 @@ pub(crate) async fn init_embedded_startup_storage_foundation(
.map_err(|err| Error::other(format!("endpoints: {err}")))?;
enforce_unsupported_fs_policy(&endpoint_pools).map_err(|err| Error::other(format!("unsupported fs guard: {err}")))?;
ecstore_global::set_global_endpoints(endpoint_pools.as_ref().clone());
ecstore_global::update_erasure_type(setup_type).await;
set_global_endpoints(endpoint_pools.as_ref().clone());
update_erasure_type(setup_type).await;
ecstore_storage::init_local_disks(endpoint_pools.clone())
init_local_disks(endpoint_pools.clone())
.await
.map_err(|err| Error::other(format!("local disks: {err}")))?;
ecstore_storage::init_lock_clients(endpoint_pools.clone());
init_lock_clients(endpoint_pools.clone());
Ok(endpoint_pools)
}
@@ -159,7 +158,7 @@ pub(crate) async fn init_startup_storage_runtime(
init_startup_storage_global_config(store.clone()).await?;
readiness.mark_stage(SystemStage::StorageReady);
ecstore_replication::init_background_replication(store.clone()).await;
init_background_replication(store.clone()).await;
Ok(StartupStorageRuntime {
store,
@@ -190,17 +189,17 @@ pub(crate) async fn init_embedded_startup_storage_runtime(
init_embedded_startup_storage_global_config(store.clone()).await?;
readiness.mark_stage(SystemStage::StorageReady);
ecstore_replication::init_background_replication(store.clone()).await;
init_background_replication(store.clone()).await;
Ok(StartupStorageRuntime { store, shutdown_token })
}
async fn init_startup_storage_global_config(store: Arc<ECStore>) -> Result<()> {
ecstore_config::init();
ecstore_config::try_migrate_server_config(store.clone()).await;
init_ecstore_config();
try_migrate_server_config(store.clone()).await;
let mut retry_count = 0;
while let Err(e) = ecstore_config::init_global_config_sys(store.clone()).await {
while let Err(e) = init_global_config_sys(store.clone()).await {
let next_retry_count = retry_count + 1;
error!(
target: "rustfs::main::run",
@@ -225,11 +224,11 @@ async fn init_startup_storage_global_config(store: Arc<ECStore>) -> Result<()> {
}
async fn init_embedded_startup_storage_global_config(store: Arc<ECStore>) -> Result<()> {
ecstore_config::init();
ecstore_config::try_migrate_server_config(store.clone()).await;
init_ecstore_config();
try_migrate_server_config(store.clone()).await;
let mut retry = 0;
while let Err(err) = ecstore_config::init_global_config_sys(store.clone()).await {
while let Err(err) = init_global_config_sys(store.clone()).await {
retry += 1;
if retry > GLOBAL_CONFIG_INIT_MAX_RETRIES {
return Err(Error::other(format!(
rustfs/src/storage/ecstore_compat.rs
+142
View File
@@ -0,0 +1,142 @@
// 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.
pub(crate) mod ecstore_admin {
pub(crate) use rustfs_ecstore::api::admin::{get_local_server_property, get_server_info};
}
pub(crate) mod ecstore_bucket {
pub(crate) use rustfs_ecstore::api::bucket::{
bandwidth, bucket_target_sys, lifecycle, metadata, metadata_sys, migration, object_lock, policy_sys, replication,
tagging, target, utils,
};
pub(crate) use rustfs_ecstore::api::bucket::{quota, versioning, versioning_sys};
}
pub(crate) mod ecstore_capacity {
pub(crate) use rustfs_ecstore::api::capacity::{
PoolDecommissionInfo, PoolStatus, get_total_usable_capacity, get_total_usable_capacity_free,
is_reserved_or_invalid_bucket,
};
}
pub(crate) mod ecstore_client {
pub(crate) use rustfs_ecstore::api::client::{admin_handler_utils, object_api_utils, transition_api};
}
pub(crate) mod ecstore_compression {
pub(crate) use rustfs_ecstore::api::compression::{MIN_DISK_COMPRESSIBLE_SIZE, is_disk_compressible};
}
pub(crate) mod ecstore_cluster {
pub(crate) use rustfs_ecstore::api::cluster::topology_snapshot_from_endpoint_pools_with_capabilities;
}
pub(crate) mod ecstore_config {
pub(crate) use rustfs_ecstore::api::config::{
com, init, init_global_config_sys, set_global_storage_class, storageclass, try_migrate_server_config,
};
}
pub(crate) mod ecstore_data_usage {
pub(crate) use rustfs_ecstore::api::data_usage::{
apply_bucket_usage_memory_overlay, load_data_usage_from_backend, record_bucket_object_delete_memory,
record_bucket_object_write_memory,
};
}
#[allow(unused_imports)]
pub(crate) mod ecstore_disk {
pub(crate) use rustfs_ecstore::api::disk::{
CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskStore, FileInfoVersions, FileReader, FileWriter,
RUSTFS_META_BUCKET, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, UpdateMetadataOpts, VolumeInfo,
WalkDirOptions,
};
pub(crate) use rustfs_ecstore::api::disk::{endpoint, error, error_reduce};
}
pub(crate) mod ecstore_error {
pub(crate) use rustfs_ecstore::api::error::{
Error, Result, StorageError, is_err_bucket_not_found, is_err_object_not_found, is_err_version_not_found,
};
}
pub(crate) mod ecstore_event {
pub(crate) use rustfs_ecstore::api::event::{EventArgs, register_event_dispatch_hook};
}
pub(crate) mod ecstore_global {
pub(crate) use rustfs_ecstore::api::global::{
GLOBAL_BOOT_TIME, GLOBAL_TierConfigMgr, get_global_bucket_monitor, get_global_deployment_id, get_global_endpoints_opt,
get_global_lock_client, get_global_lock_clients, get_global_region, get_global_tier_config_mgr, global_rustfs_port,
is_dist_erasure, new_object_layer_fn, resolve_object_store_handle, set_global_endpoints, set_global_region,
set_global_rustfs_port, set_object_store_resolver, shutdown_background_services, update_erasure_type,
};
}
#[allow(unused_imports)]
pub(crate) mod ecstore_layout {
pub(crate) use rustfs_ecstore::api::layout::{DisksLayout, EndpointServerPools, Endpoints, PoolEndpoints, SetupType};
}
pub(crate) mod ecstore_metrics {
pub(crate) use rustfs_ecstore::api::metrics::{CollectMetricsOpts, MetricType, collect_local_metrics};
}
#[allow(unused_imports)]
pub(crate) mod ecstore_notification {
pub(crate) use rustfs_ecstore::api::notification::{
NotificationSys, get_global_notification_sys, new_global_notification_sys,
};
}
#[allow(unused_imports)]
pub(crate) mod ecstore_rebalance {
pub(crate) use rustfs_ecstore::api::rebalance::{
DiskStat, RebalSaveOpt, RebalStatus, RebalanceCleanupWarningEntry, RebalanceCleanupWarnings, RebalanceInfo,
RebalanceMeta, RebalanceStats, RebalanceStopPropagationRecord, decode_rebalance_stop_propagation_record,
encode_rebalance_stop_propagation_record,
};
}
pub(crate) mod ecstore_rio {
#[cfg(test)]
pub(crate) use rustfs_ecstore::api::rio::{DecryptReader, EncryptReader, HardLimitReader, Reader, boxed_reader};
pub(crate) use rustfs_ecstore::api::rio::{
DynReader, HashReader, ReadStream, WriteEncryption, WritePlan, compression_metadata_value, wrap_reader,
};
}
pub(crate) mod ecstore_rpc {
pub(crate) use rustfs_ecstore::api::rpc::{
LocalPeerS3Client, PEER_RESTSIGNAL, PEER_RESTSUB_SYS, PeerRestClient, PeerS3Client, SERVICE_SIGNAL_REFRESH_CONFIG,
SERVICE_SIGNAL_RELOAD_DYNAMIC, TONIC_RPC_PREFIX, verify_rpc_signature,
};
}
pub(crate) mod ecstore_set_disk {
pub(crate) use rustfs_ecstore::api::set_disk::{DEFAULT_READ_BUFFER_SIZE, get_lock_acquire_timeout, is_valid_storage_class};
}
pub(crate) mod ecstore_storage {
pub(crate) use rustfs_ecstore::api::storage::{
ECStore, all_local_disk, all_local_disk_path, find_local_disk_by_ref, init_local_disks, init_lock_clients,
prewarm_local_disk_id_map,
};
}
pub(crate) mod ecstore_tier {
pub(crate) use rustfs_ecstore::api::tier::tier::TierConfigMgr;
pub(crate) use rustfs_ecstore::api::tier::{tier, tier_admin, tier_config, tier_handlers, warm_backend};
}
rustfs/src/storage/mod.rs
+223 -32
View File
@@ -54,42 +54,19 @@ pub(crate) use sse::{
validate_sse_headers_for_read, validate_sse_headers_for_write, validate_ssec_for_read,
};
// 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;
pub(crate) use rustfs_ecstore::api::admin as ecstore_admin;
pub(crate) use rustfs_ecstore::api::bucket as ecstore_bucket;
pub(crate) use rustfs_ecstore::api::client as ecstore_client;
pub(crate) use rustfs_ecstore::api::cluster as ecstore_cluster;
pub(crate) use rustfs_ecstore::api::config as ecstore_config;
pub(crate) use rustfs_ecstore::api::disk as ecstore_disk;
pub(crate) use rustfs_ecstore::api::error as ecstore_error;
pub(crate) use rustfs_ecstore::api::event as ecstore_event;
pub(crate) use rustfs_ecstore::api::global as ecstore_global;
pub(crate) use rustfs_ecstore::api::layout as ecstore_layout;
pub(crate) use rustfs_ecstore::api::metrics as ecstore_metrics;
pub(crate) use rustfs_ecstore::api::notification as ecstore_notification;
pub(crate) use rustfs_ecstore::api::rio as ecstore_rio;
pub(crate) use rustfs_ecstore::api::rpc as ecstore_rpc;
pub(crate) use rustfs_ecstore::api::set_disk as ecstore_set_disk;
pub(crate) use rustfs_ecstore::api::storage as ecstore_storage;
pub(crate) mod ecstore_compat;
pub(crate) use ecstore_compat::*;
pub(crate) const BUCKET_ACCELERATE_CONFIG: &str = ecstore_bucket::metadata::BUCKET_ACCELERATE_CONFIG;
pub(crate) const BUCKET_LOGGING_CONFIG: &str = ecstore_bucket::metadata::BUCKET_LOGGING_CONFIG;
pub(crate) const BUCKET_REQUEST_PAYMENT_CONFIG: &str = ecstore_bucket::metadata::BUCKET_REQUEST_PAYMENT_CONFIG;
pub(crate) const BUCKET_TABLE_CATALOG_META_PREFIX: &str = ecstore_bucket::metadata::BUCKET_TABLE_CATALOG_META_PREFIX;
pub(crate) const BUCKET_TABLE_CATALOG_TABLE_BUCKETS_PREFIX: &str =
ecstore_bucket::metadata::BUCKET_TABLE_CATALOG_TABLE_BUCKETS_PREFIX;
pub(crate) const BUCKET_TABLE_CONFIG: &str = ecstore_bucket::metadata::BUCKET_TABLE_CONFIG;
pub(crate) const BUCKET_TABLE_RESERVED_PREFIX: &str = ecstore_bucket::metadata::BUCKET_TABLE_RESERVED_PREFIX;
pub(crate) const BUCKET_VERSIONING_CONFIG: &str = ecstore_bucket::metadata::BUCKET_VERSIONING_CONFIG;
pub(crate) const BUCKET_WEBSITE_CONFIG: &str = ecstore_bucket::metadata::BUCKET_WEBSITE_CONFIG;
pub(crate) const DEFAULT_READ_BUFFER_SIZE: usize = ecstore_set_disk::DEFAULT_READ_BUFFER_SIZE;
@@ -98,6 +75,8 @@ pub(crate) const PEER_RESTSIGNAL: &str = ecstore_rpc::PEER_RESTSIGNAL;
pub(crate) const PEER_RESTSUB_SYS: &str = ecstore_rpc::PEER_RESTSUB_SYS;
pub(crate) const SERVICE_SIGNAL_REFRESH_CONFIG: u64 = ecstore_rpc::SERVICE_SIGNAL_REFRESH_CONFIG;
pub(crate) const SERVICE_SIGNAL_RELOAD_DYNAMIC: u64 = ecstore_rpc::SERVICE_SIGNAL_RELOAD_DYNAMIC;
pub(crate) const RUSTFS_META_BUCKET: &str = ecstore_disk::RUSTFS_META_BUCKET;
pub(crate) const TONIC_RPC_PREFIX: &str = ecstore_rpc::TONIC_RPC_PREFIX;
#[cfg(test)]
pub(crate) const STORAGE_CLASS_SUB_SYS: &str = ecstore_config::com::STORAGE_CLASS_SUB_SYS;
@@ -113,32 +92,161 @@ pub(crate) type DiskInfo = ecstore_disk::DiskInfo;
pub(crate) type DiskInfoOptions = ecstore_disk::DiskInfoOptions;
pub(crate) type DiskResult<T> = ecstore_disk::error::Result<T>;
pub(crate) type DiskStore = ecstore_disk::DiskStore;
#[cfg(test)]
pub(crate) type DisksLayout = ecstore_layout::DisksLayout;
pub(crate) type DynReplicationPool = ecstore_bucket::replication::DynReplicationPool;
pub(crate) type DynReader = ecstore_rio::DynReader;
pub(crate) type ECStore = ecstore_storage::ECStore;
pub(crate) type Endpoint = ecstore_disk::endpoint::Endpoint;
#[cfg(test)]
pub(crate) type Endpoints = ecstore_layout::Endpoints;
pub(crate) type EndpointServerPools = ecstore_layout::EndpointServerPools;
pub(crate) type EventArgs = ecstore_event::EventArgs;
pub(crate) type FileInfoVersions = ecstore_disk::FileInfoVersions;
pub(crate) type FileReader = ecstore_disk::FileReader;
pub(crate) type FileWriter = ecstore_disk::FileWriter;
pub(crate) type HashReader = ecstore_rio::HashReader;
pub(crate) type LocalPeerS3Client = ecstore_rpc::LocalPeerS3Client;
pub(crate) type MetricType = ecstore_metrics::MetricType;
pub(crate) type ObjectPartInfo = rustfs_filemeta::ObjectPartInfo;
pub(crate) type ObjectLockBlockReason = ecstore_bucket::object_lock::objectlock_sys::ObjectLockBlockReason;
pub(crate) type ObjectStoreResolver = dyn Fn() -> Option<Arc<ECStore>> + Send + Sync + 'static;
pub(crate) type PolicySys = ecstore_bucket::policy_sys::PolicySys;
pub(crate) type PoolEndpoints = ecstore_layout::PoolEndpoints;
pub(crate) type QuotaError = ecstore_bucket::quota::QuotaError;
pub(crate) type RawFileInfo = rustfs_filemeta::RawFileInfo;
pub(crate) type ReadMultipleReq = ecstore_disk::ReadMultipleReq;
pub(crate) type ReadMultipleResp = ecstore_disk::ReadMultipleResp;
pub(crate) type ReadOptions = ecstore_disk::ReadOptions;
pub(crate) type RenameDataResp = ecstore_disk::RenameDataResp;
pub(crate) type SetupType = ecstore_layout::SetupType;
pub(crate) type StorageError = ecstore_error::StorageError;
pub(crate) type Error = StorageError;
pub(crate) type Result<T> = core::result::Result<T, Error>;
pub(crate) type TierConfigMgr = ecstore_tier::TierConfigMgr;
pub(crate) type Error = ecstore_error::Error;
pub(crate) type Result<T> = ecstore_error::Result<T>;
pub(crate) type UpdateMetadataOpts = ecstore_disk::UpdateMetadataOpts;
pub(crate) type VolumeInfo = ecstore_disk::VolumeInfo;
pub(crate) type WalkDirOptions = ecstore_disk::WalkDirOptions;
pub(crate) type WriteEncryption = ecstore_rio::WriteEncryption;
pub(crate) type WritePlan = ecstore_rio::WritePlan;
#[cfg(test)]
pub(crate) type DecryptReader<R> = ecstore_rio::DecryptReader<R>;
#[cfg(test)]
pub(crate) type EncryptReader<R> = ecstore_rio::EncryptReader<R>;
#[cfg(test)]
pub(crate) type HardLimitReader<R> = ecstore_rio::HardLimitReader<R>;
pub(crate) type NotificationSys = ecstore_notification::NotificationSys;
pub(crate) async fn get_local_server_property() -> rustfs_madmin::ServerProperties {
ecstore_admin::get_local_server_property().await
}
pub(crate) async fn init_background_replication(store: Arc<ECStore>) {
ecstore_bucket::replication::init_background_replication(store).await;
}
pub(crate) async fn all_local_disk() -> Vec<DiskStore> {
ecstore_storage::all_local_disk().await
}
pub(crate) async fn get_bucket_notification_config(bucket: &str) -> Result<Option<s3s::dto::NotificationConfiguration>> {
ecstore_bucket::metadata_sys::get_notification_config(bucket).await
}
pub(crate) async fn init_bucket_metadata_sys(api: Arc<ECStore>, buckets: Vec<String>) {
ecstore_bucket::metadata_sys::init_bucket_metadata_sys(api, buckets).await;
}
pub(crate) fn bucket_metadata_runtime_initialized() -> bool {
ecstore_bucket::metadata_sys::get_global_bucket_metadata_sys().is_some()
}
pub(crate) fn disk_drive_path(disk: &DiskStore) -> String {
ecstore_disk::DiskAPI::to_string(disk.as_ref())
}
pub(crate) fn disk_endpoint(disk: &DiskStore) -> String {
ecstore_disk::DiskAPI::endpoint(disk.as_ref()).to_string()
}
pub(crate) fn get_global_replication_pool() -> Option<Arc<DynReplicationPool>> {
ecstore_bucket::replication::get_global_replication_pool()
}
pub(crate) async fn try_migrate_bucket_metadata(store: Arc<ECStore>) {
ecstore_bucket::migration::try_migrate_bucket_metadata(store).await;
}
pub(crate) async fn try_migrate_iam_config(store: Arc<ECStore>) {
ecstore_bucket::migration::try_migrate_iam_config(store).await;
}
pub(crate) fn init_ecstore_config() {
ecstore_config::init();
}
pub(crate) async fn init_global_config_sys(store: Arc<ECStore>) -> Result<()> {
ecstore_config::init_global_config_sys(store).await
}
pub(crate) async fn init_local_disks(endpoint_pools: EndpointServerPools) -> Result<()> {
ecstore_storage::init_local_disks(endpoint_pools).await
}
pub(crate) fn init_lock_clients(endpoint_pools: EndpointServerPools) {
ecstore_storage::init_lock_clients(endpoint_pools);
}
pub(crate) async fn new_global_notification_sys(endpoint_pools: EndpointServerPools) -> Result<()> {
ecstore_notification::new_global_notification_sys(endpoint_pools).await
}
pub(crate) async fn read_config(api: Arc<ECStore>, file: &str) -> Result<Vec<u8>> {
ecstore_config::com::read_config(api, file).await
}
pub(crate) async fn prewarm_local_disk_id_map() {
ecstore_storage::prewarm_local_disk_id_map().await;
}
pub(crate) fn replication_queue_current_count() -> Option<i64> {
ecstore_bucket::replication::GLOBAL_REPLICATION_STATS.get().and_then(|stats| {
stats
.q_cache
.try_lock()
.ok()
.map(|cache| cache.sr_queue_stats.curr.get_current_count())
})
}
pub(crate) async fn save_config(api: Arc<ECStore>, file: &str, data: Vec<u8>) -> Result<()> {
ecstore_config::com::save_config(api, file, data).await
}
pub(crate) fn shutdown_background_services() {
ecstore_global::shutdown_background_services();
}
pub(crate) fn set_global_endpoints(endpoints: Vec<PoolEndpoints>) {
ecstore_global::set_global_endpoints(endpoints);
}
pub(crate) fn set_global_region(region: s3s::region::Region) {
ecstore_global::set_global_region(region);
}
pub(crate) fn set_global_rustfs_port(value: u16) {
ecstore_global::set_global_rustfs_port(value);
}
pub(crate) async fn try_migrate_server_config(store: Arc<ECStore>) {
ecstore_config::try_migrate_server_config(store).await;
}
pub(crate) async fn update_erasure_type(setup_type: SetupType) {
ecstore_global::update_erasure_type(setup_type).await;
}
pub(crate) trait StorageDiskRpcExt {
async fn disk_info(&self, opts: &DiskInfoOptions) -> DiskResult<DiskInfo>;
async fn delete_volume(&self, volume: &str) -> DiskResult<()>;
@@ -562,14 +670,47 @@ pub(crate) fn is_err_version_not_found(err: &Error) -> bool {
ecstore_error::is_err_version_not_found(err)
}
pub(crate) fn is_all_buckets_not_found(errs: &[Option<DiskError>]) -> bool {
ecstore_disk::error_reduce::is_all_buckets_not_found(errs)
}
pub(crate) fn get_global_lock_client() -> Option<Arc<dyn rustfs_lock::client::LockClient>> {
ecstore_global::get_global_lock_client()
}
pub(crate) fn get_global_lock_clients()
-> Option<&'static std::collections::HashMap<String, Arc<dyn rustfs_lock::client::LockClient>>> {
ecstore_global::get_global_lock_clients()
}
pub(crate) fn get_global_endpoints_opt() -> Option<EndpointServerPools> {
ecstore_global::get_global_endpoints_opt()
}
pub(crate) fn get_global_region() -> Option<s3s::region::Region> {
ecstore_global::get_global_region()
}
pub(crate) fn get_global_tier_config_mgr() -> Arc<tokio::sync::RwLock<TierConfigMgr>> {
ecstore_global::get_global_tier_config_mgr()
}
pub(crate) fn new_object_layer_fn() -> Option<Arc<ECStore>> {
ecstore_global::new_object_layer_fn()
}
pub(crate) fn set_object_store_resolver(resolver: Arc<ObjectStoreResolver>) -> bool {
ecstore_global::set_object_store_resolver(resolver)
}
pub(crate) fn get_global_notification_sys() -> Option<&'static NotificationSys> {
ecstore_notification::get_global_notification_sys()
}
pub(crate) async fn is_dist_erasure() -> bool {
ecstore_global::is_dist_erasure().await
}
pub(crate) fn resolve_object_store_handle() -> Option<Arc<ECStore>> {
ecstore_global::resolve_object_store_handle()
}
@@ -585,6 +726,56 @@ pub(crate) fn verify_rpc_signature(url: &str, method: &http::Method, headers: &h
ecstore_rpc::verify_rpc_signature(url, method, headers)
}
pub(crate) fn to_s3s_etag(etag: &str) -> s3s::dto::ETag {
ecstore_client::object_api_utils::to_s3s_etag(etag)
}
pub(crate) fn table_catalog_path_hash(value: &str) -> String {
ecstore_bucket::metadata::table_catalog_path_hash(value)
}
pub(crate) fn get_lock_acquire_timeout() -> std::time::Duration {
ecstore_set_disk::get_lock_acquire_timeout()
}
#[cfg(test)]
pub(crate) fn boxed_reader<R>(reader: R) -> DynReader
where
R: ecstore_rio::Reader + 'static,
{
ecstore_rio::boxed_reader(reader)
}
pub(crate) fn compression_metadata_value(algorithm: rustfs_utils::CompressionAlgorithm) -> String {
ecstore_rio::compression_metadata_value(algorithm)
}
pub(crate) fn wrap_reader<R>(reader: R) -> DynReader
where
R: ecstore_rio::ReadStream + 'static,
{
ecstore_rio::wrap_reader(reader)
}
pub(crate) fn is_valid_storage_class(storage_class: &str) -> bool {
ecstore_set_disk::is_valid_storage_class(storage_class)
}
pub(crate) fn register_event_dispatch_hook<F>(hook: F) -> bool
where
F: Fn(EventArgs) + Send + Sync + 'static,
{
ecstore_event::register_event_dispatch_hook(hook)
}
pub(crate) fn topology_snapshot_from_endpoint_pools_with_capabilities(
endpoint_pools: &EndpointServerPools,
capabilities: rustfs_storage_api::TopologyCapabilities,
disk_capabilities: rustfs_storage_api::DiskCapabilities,
) -> rustfs_storage_api::TopologySnapshot {
ecstore_cluster::topology_snapshot_from_endpoint_pools_with_capabilities(endpoint_pools, capabilities, disk_capabilities)
}
pub(crate) async fn reload_transition_tier_config(api: Arc<ECStore>) -> std::io::Result<()> {
ecstore_global::GLOBAL_TierConfigMgr.write().await.reload(api).await
}
rustfs/src/storage/s3_api/bucket.rs
+1 -1
View File
@@ -12,8 +12,8 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use crate::storage::ecstore_client::object_api_utils::to_s3s_etag;
use crate::storage::s3_api::common::rustfs_owner;
use crate::storage::to_s3s_etag;
use percent_encoding::percent_decode_str;
use rustfs_storage_api::{
BucketInfo, ListObjectVersionsInfo as StorageListObjectVersionsInfo, ListObjectsV2Info as StorageListObjectsV2Info,
rustfs/src/storage/s3_api/multipart.rs
+2 -2
View File
@@ -12,8 +12,8 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use crate::storage::ecstore_client::object_api_utils::to_s3s_etag;
use crate::storage::s3_api::common::{rustfs_initiator, rustfs_owner};
use crate::storage::to_s3s_etag;
use rustfs_storage_api::{ListMultipartsInfo, ListPartsInfo};
use s3s::dto::{CommonPrefix, ListMultipartUploadsOutput, ListPartsOutput, MultipartUpload, Part, Timestamp};
use s3s::{S3Error, S3ErrorCode};
@@ -195,8 +195,8 @@ mod tests {
MAX_MULTIPART_UPLOADS_LIST, build_list_multipart_uploads_output, build_list_parts_output,
parse_list_multipart_uploads_params, parse_list_parts_params, parse_upload_part_number,
};
use crate::storage::ecstore_client::object_api_utils::to_s3s_etag;
use crate::storage::s3_api::common::{rustfs_initiator, rustfs_owner};
use crate::storage::to_s3s_etag;
use rustfs_storage_api::{ListMultipartsInfo, ListPartsInfo, MultipartInfo, PartInfo};
use s3s::S3ErrorCode;
use s3s::dto::Timestamp;
rustfs/src/table_catalog.rs
+22 -29
View File
@@ -28,8 +28,9 @@ use std::{
};
use crate::storage::{
ecstore_bucket::{metadata as ecstore_metadata, metadata_sys as ecstore_metadata_sys},
ecstore_disk, ecstore_error, ecstore_set_disk,
BUCKET_TABLE_CATALOG_META_PREFIX, BUCKET_TABLE_CATALOG_TABLE_BUCKETS_PREFIX, BUCKET_TABLE_CONFIG,
BUCKET_TABLE_RESERVED_PREFIX, Error as EcstoreError, RUSTFS_META_BUCKET, StorageError, get_bucket_metadata,
get_lock_acquire_timeout, table_catalog_path_hash,
};
use bytes::Bytes;
use datafusion::{
@@ -55,8 +56,7 @@ use crate::storage::{
StorageObjectOptions as ObjectOptions, StorageObjectToDelete as ObjectToDelete, StoragePutObjReader as PutObjReader,
};
pub(crate) const TABLE_BUCKET_MARKER_CONFIG: &str = ecstore_metadata::BUCKET_TABLE_CONFIG;
const RUSTFS_META_BUCKET: &str = ecstore_disk::RUSTFS_META_BUCKET;
pub(crate) const TABLE_BUCKET_MARKER_CONFIG: &str = BUCKET_TABLE_CONFIG;
pub(crate) const RESERVED_CATALOG_OBJECT_MESSAGE: &str = "Object key is reserved for the table catalog";
pub(crate) const TABLE_BUCKET_CATALOG_TYPE: &str = "iceberg-rest";
pub(crate) const TABLE_BUCKET_CONFIG_VERSION: u16 = 1;
@@ -69,7 +69,7 @@ pub(crate) const TABLE_MAINTENANCE_CONFIG_VERSION: u16 = 1;
pub(crate) const TABLE_EXTERNAL_CATALOG_BRIDGE_VERSION: u16 = 1;
pub(crate) const TABLE_CATALOG_BACKING_MANIFEST_VERSION: u16 = 1;
pub(crate) const TABLE_METADATA_FILE_NAME_MAX_LEN: usize = 128;
pub const TABLE_RESERVED_PREFIX: &str = ecstore_metadata::BUCKET_TABLE_RESERVED_PREFIX;
pub const TABLE_RESERVED_PREFIX: &str = BUCKET_TABLE_RESERVED_PREFIX;
const WAREHOUSE_ROOT: &str = "warehouses";
const NAMESPACE_ROOT: &str = "namespaces";
const TABLE_ROOT: &str = "tables";
@@ -85,8 +85,8 @@ const TABLE_BUCKET_ENTRY_FILE: &str = "table-bucket.json";
const NAMESPACE_ENTRY_FILE: &str = "namespace-entry.json";
const TABLE_ENTRY_FILE: &str = "table-entry.json";
const VIEW_ENTRY_FILE: &str = "view-entry.json";
const INTERNAL_CATALOG_ROOT: &str = ecstore_metadata::BUCKET_TABLE_CATALOG_META_PREFIX;
const TABLE_BUCKET_ROOT: &str = ecstore_metadata::BUCKET_TABLE_CATALOG_TABLE_BUCKETS_PREFIX;
const INTERNAL_CATALOG_ROOT: &str = BUCKET_TABLE_CATALOG_META_PREFIX;
const TABLE_BUCKET_ROOT: &str = BUCKET_TABLE_CATALOG_TABLE_BUCKETS_PREFIX;
const COMMIT_LOG_ROOT: &str = "commits";
const COMMIT_IDEMPOTENCY_ROOT: &str = "commit-idempotency";
const EXTERNAL_CATALOG_ROOT: &str = "external-catalog";
@@ -115,10 +115,8 @@ const ICEBERG_REF_MAX_REF_AGE_MS_FIELD: &str = "max-ref-age-ms";
type CatalogListObjectsV2Info = StorageListObjectsV2Info<ObjectInfo>;
type CatalogListObjectVersionsInfo = StorageListObjectVersionsInfo<ObjectInfo>;
type EcstoreError = ecstore_error::Error;
type CatalogObjectInfoOrErr = StorageObjectInfoOrErr<ObjectInfo, EcstoreError>;
type CatalogWalkOptions = StorageWalkOptions<fn(&FileInfo) -> bool>;
type StorageError = ecstore_error::StorageError;
pub(crate) trait TableCatalogStorage:
StorageObjectIO<
@@ -1463,7 +1461,7 @@ impl TableCatalogObjectPaths {
"{}{}/{MAINTENANCE_ROOT}/{}/{MAINTENANCE_CONFIG_FILE}",
self.table_entries_prefix(table_bucket, namespace),
table.as_str(),
ecstore_metadata::table_catalog_path_hash(table_id)
table_catalog_path_hash(table_id)
)
}
@@ -1479,8 +1477,8 @@ impl TableCatalogObjectPaths {
"{}{}/{MAINTENANCE_ROOT}/{}/{MAINTENANCE_JOB_ROOT}/{}.json",
self.table_entries_prefix(table_bucket, namespace),
table.as_str(),
ecstore_metadata::table_catalog_path_hash(table_id),
ecstore_metadata::table_catalog_path_hash(job_id)
table_catalog_path_hash(table_id),
table_catalog_path_hash(job_id)
)
}
@@ -1495,7 +1493,7 @@ impl TableCatalogObjectPaths {
"{}{}/{MAINTENANCE_ROOT}/{}/{MAINTENANCE_LATEST_JOB_FILE}",
self.table_entries_prefix(table_bucket, namespace),
table.as_str(),
ecstore_metadata::table_catalog_path_hash(table_id)
table_catalog_path_hash(table_id)
)
}
@@ -1510,7 +1508,7 @@ impl TableCatalogObjectPaths {
"{}{}/{MAINTENANCE_ROOT}/{}/{MAINTENANCE_CURRENT_JOB_FILE}",
self.table_entries_prefix(table_bucket, namespace),
table.as_str(),
ecstore_metadata::table_catalog_path_hash(table_id)
table_catalog_path_hash(table_id)
)
}
@@ -1519,8 +1517,8 @@ impl TableCatalogObjectPaths {
"{}{}/{}/{}.json",
self.table_bucket_root_prefix(table_bucket),
COMMIT_LOG_ROOT,
ecstore_metadata::table_catalog_path_hash(table_id),
ecstore_metadata::table_catalog_path_hash(commit_id)
table_catalog_path_hash(table_id),
table_catalog_path_hash(commit_id)
)
}
@@ -1529,7 +1527,7 @@ impl TableCatalogObjectPaths {
"{}{}/{}/",
self.table_bucket_root_prefix(table_bucket),
COMMIT_LOG_ROOT,
ecstore_metadata::table_catalog_path_hash(table_id)
table_catalog_path_hash(table_id)
)
}
@@ -1538,8 +1536,8 @@ impl TableCatalogObjectPaths {
"{}{}/{}/{}.json",
self.table_bucket_root_prefix(table_bucket),
COMMIT_IDEMPOTENCY_ROOT,
ecstore_metadata::table_catalog_path_hash(table_id),
ecstore_metadata::table_catalog_path_hash(idempotency_key)
table_catalog_path_hash(table_id),
table_catalog_path_hash(idempotency_key)
)
}
@@ -1548,17 +1546,12 @@ impl TableCatalogObjectPaths {
"{}{}/{}/",
self.table_bucket_root_prefix(table_bucket),
COMMIT_IDEMPOTENCY_ROOT,
ecstore_metadata::table_catalog_path_hash(table_id)
table_catalog_path_hash(table_id)
)
}
fn table_bucket_root_prefix(&self, table_bucket: &str) -> String {
format!(
"{}/{}/{}/",
self.catalog_root,
TABLE_BUCKET_ROOT,
ecstore_metadata::table_catalog_path_hash(table_bucket)
)
format!("{}/{}/{}/", self.catalog_root, TABLE_BUCKET_ROOT, table_catalog_path_hash(table_bucket))
}
}
@@ -4038,7 +4031,7 @@ where
.await
.map_err(|err| storage_error_to_catalog("create catalog table lock", err))?;
let guard = lock
.get_write_lock(ecstore_set_disk::get_lock_acquire_timeout())
.get_write_lock(get_lock_acquire_timeout())
.await
.map_err(|err| TableCatalogStoreError::Internal(format!("failed to acquire catalog table lock: {err}")))?;
Ok(Box::new(guard))
@@ -7009,7 +7002,7 @@ pub fn validate_object_mutation(table_bucket_enabled: bool, object_key: &str) ->
}
pub(crate) async fn validate_bucket_object_mutation(bucket: &str, object_key: &str) -> Result<(), TableObjectMutationError> {
let table_bucket_enabled = ecstore_metadata_sys::get(bucket)
let table_bucket_enabled = get_bucket_metadata(bucket)
.await
.is_ok_and(|metadata| metadata.table_bucket_enabled());
@@ -7417,7 +7410,7 @@ mod tests {
let bucket = "analytics";
let namespace = Namespace::parse("analytics.daily_events").unwrap();
let table = IdentifierSegment::parse("events").unwrap();
let bucket_root = format!("s3tables/catalog/table-buckets/{}/", ecstore_metadata::table_catalog_path_hash(bucket));
let bucket_root = format!("s3tables/catalog/table-buckets/{}/", table_catalog_path_hash(bucket));
assert_eq!(paths.table_bucket_entry_path(bucket), format!("{bucket_root}table-bucket.json"));
assert_eq!(
rustfs/src/workload_admission.rs
+3 -13
View File
@@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use crate::storage::ecstore_bucket::{metadata_sys, replication};
use crate::storage::{bucket_metadata_runtime_initialized, get_global_replication_pool, replication_queue_current_count};
use rustfs_concurrency::{
AdmissionState, WorkloadAdmissionRegistrySnapshot, WorkloadAdmissionSnapshot, WorkloadAdmissionSnapshotProvider,
WorkloadClass,
@@ -75,7 +75,7 @@ pub fn foreground_read_workload_admission_snapshot() -> WorkloadAdmissionSnapsho
}
pub fn metadata_workload_admission_snapshot() -> WorkloadAdmissionSnapshot {
metadata_workload_admission_snapshot_from_initialized(metadata_sys::get_global_bucket_metadata_sys().is_some())
metadata_workload_admission_snapshot_from_initialized(bucket_metadata_runtime_initialized())
}
fn metadata_workload_admission_snapshot_from_initialized(runtime_initialized: bool) -> WorkloadAdmissionSnapshot {
@@ -151,7 +151,7 @@ fn repair_workload_admission_snapshot_from_counts(
}
pub fn replication_workload_admission_snapshot() -> WorkloadAdmissionSnapshot {
let Some(pool) = replication::get_global_replication_pool() else {
let Some(pool) = get_global_replication_pool() else {
return replication_workload_admission_snapshot_from_counts(false, None, None);
};
@@ -198,16 +198,6 @@ fn i32_to_usize_saturated(value: i32) -> usize {
usize::try_from(value.max(0)).unwrap_or(usize::MAX)
}
fn replication_queue_current_count() -> Option<i64> {
replication::GLOBAL_REPLICATION_STATS.get().and_then(|stats| {
stats
.q_cache
.try_lock()
.ok()
.map(|cache| cache.sr_queue_stats.curr.get_current_count())
})
}
#[cfg(test)]
mod tests {
use super::*;