mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-09 14:49:25 +00:00
refactor: remove remaining compatibility bridges (#3738)
This commit is contained in:
@@ -15,10 +15,10 @@
|
||||
#[cfg(test)]
|
||||
#[allow(unsafe_op_in_unsafe_fn)]
|
||||
mod tests {
|
||||
use super::super::super::config_storage_compat::DisksLayout;
|
||||
use crate::config::{CommandResult, Config, Opt, TlsCommands};
|
||||
use rustfs_config::{DEFAULT_CONSOLE_ADDRESS, DEFAULT_CONSOLE_ENABLE, DEFAULT_OBS_ENDPOINT, RUSTFS_REGION};
|
||||
use rustfs_credentials::{DEFAULT_ACCESS_KEY, DEFAULT_SECRET_KEY};
|
||||
use rustfs_ecstore::api::layout::DisksLayout;
|
||||
use serial_test::serial;
|
||||
use std::env;
|
||||
|
||||
@@ -262,7 +262,7 @@ mod tests {
|
||||
#[test]
|
||||
#[serial]
|
||||
fn test_volumes_and_disk_layout_parsing() {
|
||||
use super::super::super::config_storage_compat::DisksLayout;
|
||||
use rustfs_ecstore::api::layout::DisksLayout;
|
||||
|
||||
// Test case 1: Single volume path
|
||||
let args = vec!["rustfs", "/data/vol1"];
|
||||
|
||||
@@ -1,19 +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.
|
||||
|
||||
#[cfg(test)]
|
||||
use rustfs_ecstore::api::layout as ecstore_layout;
|
||||
|
||||
#[cfg(test)]
|
||||
pub(crate) type DisksLayout = ecstore_layout::DisksLayout;
|
||||
+1
-1
@@ -12,7 +12,7 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use super::error_storage_compat::{QuotaError, StorageError};
|
||||
use rustfs_ecstore::api::{bucket::quota::QuotaError, error::StorageError};
|
||||
use rustfs_storage_api::HTTPRangeError;
|
||||
use s3s::{S3Error, S3ErrorCode};
|
||||
|
||||
|
||||
@@ -1,19 +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::bucket as ecstore_bucket;
|
||||
use rustfs_ecstore::api::error as ecstore_error;
|
||||
|
||||
pub(crate) type QuotaError = ecstore_bucket::quota::QuotaError;
|
||||
pub(crate) type StorageError = ecstore_error::StorageError;
|
||||
+16
-14
@@ -12,7 +12,6 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use super::startup_storage_compat::{get_global_region, get_notification_config};
|
||||
use crate::server::ShutdownHandle;
|
||||
use crate::storage::{process_lambda_configurations, process_queue_configurations, process_topic_configurations};
|
||||
use crate::{admin, config, version};
|
||||
@@ -20,6 +19,7 @@ use rustfs_config::{
|
||||
DEFAULT_BUFFER_MAX_SIZE, DEFAULT_BUFFER_MIN_SIZE, DEFAULT_BUFFER_PROFILE, DEFAULT_BUFFER_UNKNOWN_SIZE, DEFAULT_UPDATE_CHECK,
|
||||
ENV_RUSTFS_BUFFER_DEFAULT_SIZE, ENV_RUSTFS_BUFFER_MAX_SIZE, ENV_RUSTFS_BUFFER_MIN_SIZE, ENV_UPDATE_CHECK, RUSTFS_REGION,
|
||||
};
|
||||
use rustfs_ecstore::api::{bucket::metadata_sys as ecstore_metadata_sys, global as ecstore_global};
|
||||
use rustfs_notify::notifier_global;
|
||||
use rustfs_targets::arn::{ARN, TargetIDError};
|
||||
use rustfs_utils::get_env_usize;
|
||||
@@ -157,7 +157,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 = get_global_region();
|
||||
let global_region = ecstore_global::get_global_region();
|
||||
let region = global_region
|
||||
.as_ref()
|
||||
.filter(|r| !r.as_str().is_empty())
|
||||
@@ -174,18 +174,20 @@ pub async fn add_bucket_notification_configuration(buckets: Vec<String>) {
|
||||
RUSTFS_REGION
|
||||
});
|
||||
for bucket in buckets.iter() {
|
||||
let has_notification_config = 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 = 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
|
||||
});
|
||||
|
||||
match has_notification_config {
|
||||
Some(cfg) => {
|
||||
|
||||
@@ -57,11 +57,9 @@ pub mod auth;
|
||||
pub mod auth_keystone;
|
||||
pub mod capacity;
|
||||
pub mod config;
|
||||
pub(crate) mod config_storage_compat;
|
||||
pub mod delete_tail_activity;
|
||||
pub mod embedded;
|
||||
pub mod error;
|
||||
pub(crate) mod error_storage_compat;
|
||||
pub mod init;
|
||||
pub mod license;
|
||||
pub mod memory_observability;
|
||||
@@ -69,7 +67,6 @@ pub mod profiling;
|
||||
#[cfg(any(feature = "ftps", feature = "webdav", feature = "sftp"))]
|
||||
pub mod protocols;
|
||||
pub mod runtime_capabilities;
|
||||
pub(crate) mod runtime_capabilities_storage_compat;
|
||||
pub mod server;
|
||||
pub(crate) mod startup_audit;
|
||||
pub(crate) mod startup_auth;
|
||||
@@ -93,16 +90,13 @@ pub(crate) mod startup_server;
|
||||
pub(crate) mod startup_services;
|
||||
pub(crate) mod startup_shutdown;
|
||||
pub(crate) mod startup_storage;
|
||||
pub(crate) mod startup_storage_compat;
|
||||
pub(crate) mod startup_tls_material;
|
||||
pub mod storage;
|
||||
pub(crate) mod table_catalog;
|
||||
pub(crate) mod table_catalog_storage_compat;
|
||||
pub mod tls;
|
||||
pub mod update;
|
||||
pub mod version;
|
||||
pub mod workload_admission;
|
||||
pub(crate) mod workload_admission_storage_compat;
|
||||
|
||||
// Re-export from rustfs_utils so that config sub-modules can use
|
||||
// `crate::apply_external_env_compat` without breaking.
|
||||
|
||||
@@ -12,14 +12,13 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use rustfs_ecstore::api::{cluster as ecstore_cluster, layout::EndpointServerPools};
|
||||
use rustfs_storage_api::{
|
||||
CapabilitySnapshotError, CapabilityStatus, DiskCapabilities, MemorySamplingState, ObservabilitySnapshot,
|
||||
ObservabilitySnapshotProvider, PlatformSupport, TopologyCapabilities, TopologySnapshot, TopologySnapshotProvider,
|
||||
UserspaceProfilingCapability,
|
||||
};
|
||||
|
||||
use super::runtime_capabilities_storage_compat::{EndpointServerPools, topology_snapshot_from_endpoint_pools_with_capabilities};
|
||||
|
||||
const NOT_WIRED_INTO_RUNTIME: &str = "not wired into runtime";
|
||||
const STORAGE_MEDIA_NOT_REPORTED: &str = "storage media not reported by endpoints";
|
||||
const FAILURE_DOMAIN_NOT_REPORTED: &str = "failure domain labels not reported by endpoints";
|
||||
@@ -80,7 +79,7 @@ impl TopologySnapshotProvider for EndpointTopologySnapshotProvider {
|
||||
}
|
||||
|
||||
pub fn topology_snapshot_from_endpoint_pools(endpoint_pools: &EndpointServerPools) -> TopologySnapshot {
|
||||
topology_snapshot_from_endpoint_pools_with_capabilities(
|
||||
ecstore_cluster::topology_snapshot_from_endpoint_pools_with_capabilities(
|
||||
endpoint_pools,
|
||||
TopologyCapabilities {
|
||||
profiling: cpu_profiling_status(),
|
||||
@@ -131,8 +130,11 @@ fn cgroup_memory_status() -> CapabilityStatus {
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::super::runtime_capabilities_storage_compat::{Endpoint, Endpoints, PoolEndpoints};
|
||||
use super::*;
|
||||
use rustfs_ecstore::api::{
|
||||
disk::endpoint::Endpoint,
|
||||
layout::{Endpoints, PoolEndpoints},
|
||||
};
|
||||
use rustfs_storage_api::{CapabilityState, ObservabilitySnapshotProvider, TopologySnapshotProvider};
|
||||
|
||||
#[tokio::test]
|
||||
|
||||
@@ -1,34 +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::cluster as ecstore_cluster;
|
||||
#[cfg(test)]
|
||||
use rustfs_ecstore::api::disk as ecstore_disk;
|
||||
use rustfs_ecstore::api::layout as ecstore_layout;
|
||||
|
||||
#[cfg(test)]
|
||||
pub(crate) type Endpoint = ecstore_disk::endpoint::Endpoint;
|
||||
pub(crate) type EndpointServerPools = ecstore_layout::EndpointServerPools;
|
||||
#[cfg(test)]
|
||||
pub(crate) type Endpoints = ecstore_layout::Endpoints;
|
||||
#[cfg(test)]
|
||||
pub(crate) type PoolEndpoints = ecstore_layout::PoolEndpoints;
|
||||
|
||||
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)
|
||||
}
|
||||
@@ -12,7 +12,7 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use super::startup_storage_compat::ECStore;
|
||||
use rustfs_ecstore::api::storage::ECStore;
|
||||
use rustfs_heal::{create_ahm_services_cancel_token, heal::storage::ECStoreHealStorage, init_heal_manager};
|
||||
use rustfs_utils::get_env_bool_with_aliases;
|
||||
use std::{io::Result, sync::Arc};
|
||||
|
||||
@@ -12,8 +12,9 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use super::startup_storage_compat::{
|
||||
ECStore, get_global_replication_pool, init_bucket_metadata_sys, try_migrate_bucket_metadata, try_migrate_iam_config,
|
||||
use rustfs_ecstore::api::{
|
||||
bucket::{metadata_sys as ecstore_metadata_sys, migration as ecstore_migration, replication as ecstore_replication},
|
||||
storage::ECStore,
|
||||
};
|
||||
use rustfs_storage_api::{BucketOperations, BucketOptions};
|
||||
use std::{
|
||||
@@ -33,9 +34,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();
|
||||
|
||||
try_migrate_bucket_metadata(store.clone()).await;
|
||||
init_bucket_metadata_sys(store.clone(), buckets.clone()).await;
|
||||
try_migrate_iam_config(store).await;
|
||||
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;
|
||||
|
||||
Ok(buckets)
|
||||
}
|
||||
@@ -51,14 +52,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();
|
||||
|
||||
try_migrate_bucket_metadata(store.clone()).await;
|
||||
ecstore_migration::try_migrate_bucket_metadata(store.clone()).await;
|
||||
|
||||
if let Some(pool) = get_global_replication_pool() {
|
||||
if let Some(pool) = ecstore_replication::get_global_replication_pool() {
|
||||
pool.init_resync(ctx, buckets.clone()).await?;
|
||||
}
|
||||
|
||||
try_migrate_iam_config(store.clone()).await;
|
||||
init_bucket_metadata_sys(store, buckets.clone()).await;
|
||||
ecstore_migration::try_migrate_iam_config(store.clone()).await;
|
||||
ecstore_metadata_sys::init_bucket_metadata_sys(store, buckets.clone()).await;
|
||||
|
||||
Ok(buckets)
|
||||
}
|
||||
|
||||
@@ -12,11 +12,11 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use super::startup_storage_compat::EndpointServerPools;
|
||||
use rustfs_config::{
|
||||
DEFAULT_RUSTFS_UNSUPPORTED_FS_POLICY, ENV_RUSTFS_UNSUPPORTED_FS_POLICY, RUSTFS_UNSUPPORTED_FS_POLICY_FAIL,
|
||||
RUSTFS_UNSUPPORTED_FS_POLICY_WARN,
|
||||
};
|
||||
use rustfs_ecstore::api::layout::EndpointServerPools;
|
||||
use std::collections::BTreeSet;
|
||||
use std::io::{Error, Result};
|
||||
use tracing::warn;
|
||||
|
||||
@@ -12,10 +12,10 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use super::startup_storage_compat::ECStore;
|
||||
use crate::app::context::AppContext;
|
||||
use crate::server::{ServiceStateManager, publish_ready_when_runtime_ready};
|
||||
use rustfs_common::{GlobalReadiness, SystemStage};
|
||||
use rustfs_ecstore::api::storage::ECStore;
|
||||
use rustfs_iam::init_iam_sys;
|
||||
use rustfs_kms::KmsServiceManager;
|
||||
use std::future::Future;
|
||||
|
||||
@@ -17,9 +17,9 @@ use crate::{
|
||||
startup_iam::{IamBootstrapDisposition, publish_ready_for_iam_bootstrap},
|
||||
startup_services::StartupServiceRuntime,
|
||||
startup_shutdown::run_startup_shutdown_sequence,
|
||||
startup_storage_compat::ECStore,
|
||||
};
|
||||
use rustfs_common::GlobalReadiness;
|
||||
use rustfs_ecstore::api::storage::ECStore;
|
||||
use rustfs_scanner::init_data_scanner;
|
||||
use std::{
|
||||
io::Result,
|
||||
|
||||
@@ -12,8 +12,8 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use super::startup_storage_compat::{EcstoreResult, EndpointServerPools, new_global_notification_sys};
|
||||
use crate::init::add_bucket_notification_configuration;
|
||||
use rustfs_ecstore::api::{error as ecstore_error, layout::EndpointServerPools, notification as ecstore_notification};
|
||||
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) -> EcstoreResult<()> {
|
||||
init_notification_system_with(|| new_global_notification_sys(endpoint_pools)).await
|
||||
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
|
||||
}
|
||||
|
||||
async fn init_notification_system_with<InitFn, InitFuture>(init_notification: InitFn) -> EcstoreResult<()>
|
||||
async fn init_notification_system_with<InitFn, InitFuture>(init_notification: InitFn) -> ecstore_error::Result<()>
|
||||
where
|
||||
InitFn: FnOnce() -> InitFuture,
|
||||
InitFuture: Future<Output = EcstoreResult<()>>,
|
||||
InitFuture: Future<Output = ecstore_error::Result<()>>,
|
||||
{
|
||||
init_notification().await
|
||||
}
|
||||
@@ -76,11 +76,11 @@ fn log_embedded_optional_service_skipped(service: &str, err: impl std::fmt::Disp
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::init_notification_system_with;
|
||||
use rustfs_ecstore::api::error::Error as EcstoreError;
|
||||
|
||||
#[tokio::test]
|
||||
async fn notification_system_returns_source_error() {
|
||||
let result =
|
||||
init_notification_system_with(|| async { Err(super::super::startup_storage_compat::EcstoreError::FaultyDisk) }).await;
|
||||
let result = init_notification_system_with(|| async { Err(EcstoreError::FaultyDisk) }).await;
|
||||
|
||||
assert!(result.is_err());
|
||||
}
|
||||
|
||||
@@ -12,7 +12,6 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use super::startup_storage_compat::{set_global_region, set_global_rustfs_port};
|
||||
use crate::{
|
||||
capacity::capacity_integration::init_capacity_management,
|
||||
config::Config,
|
||||
@@ -20,6 +19,7 @@ use crate::{
|
||||
};
|
||||
use rustfs_common::{GlobalReadiness, set_global_addr};
|
||||
use rustfs_credentials::init_global_action_credentials;
|
||||
use rustfs_ecstore::api::global::{set_global_region, set_global_rustfs_port};
|
||||
use rustfs_utils::net::parse_and_resolve_address;
|
||||
use std::{
|
||||
io::{Error, Result},
|
||||
|
||||
@@ -12,7 +12,6 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use super::startup_storage_compat::{ECStore, EndpointServerPools};
|
||||
use crate::{
|
||||
config::Config,
|
||||
init::{init_buffer_profile_system, init_kms_system},
|
||||
@@ -29,6 +28,7 @@ use crate::{
|
||||
startup_optional_runtime_sidecars::{OptionalRuntimeServices, init_optional_runtime_services},
|
||||
};
|
||||
use rustfs_common::GlobalReadiness;
|
||||
use rustfs_ecstore::api::{layout::EndpointServerPools, storage::ECStore};
|
||||
use std::{io::Result, sync::Arc};
|
||||
use tokio_util::sync::CancellationToken;
|
||||
|
||||
|
||||
@@ -17,8 +17,8 @@ use crate::{
|
||||
startup_optional_runtime_sidecars::{
|
||||
OptionalRuntimeServices, prepare_optional_runtime_shutdowns, shutdown_optional_runtime_services,
|
||||
},
|
||||
startup_storage_compat::shutdown_background_services,
|
||||
};
|
||||
use rustfs_ecstore::api::global::shutdown_background_services;
|
||||
use rustfs_heal::shutdown_ahm_services;
|
||||
use rustfs_utils::get_env_bool_with_aliases;
|
||||
use std::path::Path;
|
||||
|
||||
@@ -12,12 +12,13 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use super::startup_storage_compat::{
|
||||
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 crate::startup_fs_guard::enforce_unsupported_fs_policy;
|
||||
use rustfs_common::{GlobalReadiness, SystemStage};
|
||||
use rustfs_ecstore::api::storage::ECStore;
|
||||
use rustfs_ecstore::api::{
|
||||
bucket::replication as ecstore_replication, config as ecstore_config, global as ecstore_global, layout::EndpointServerPools,
|
||||
storage as ecstore_storage,
|
||||
};
|
||||
use std::{
|
||||
io::{Error, Result},
|
||||
net::SocketAddr,
|
||||
@@ -70,8 +71,8 @@ pub(crate) async fn init_startup_storage_foundation(server_address: &str, volume
|
||||
.map_err(Error::other)?;
|
||||
enforce_unsupported_fs_policy(&endpoint_pools)?;
|
||||
|
||||
set_global_endpoints(endpoint_pools.as_ref().clone());
|
||||
update_erasure_type(setup_type).await;
|
||||
ecstore_global::set_global_endpoints(endpoint_pools.as_ref().clone());
|
||||
ecstore_global::update_erasure_type(setup_type).await;
|
||||
|
||||
debug!(
|
||||
target: "rustfs::main::run",
|
||||
@@ -82,7 +83,7 @@ pub(crate) async fn init_startup_storage_foundation(server_address: &str, volume
|
||||
state = "starting",
|
||||
"starting local disk initialization"
|
||||
);
|
||||
init_local_disks(endpoint_pools.clone())
|
||||
ecstore_storage::init_local_disks(endpoint_pools.clone())
|
||||
.await
|
||||
.inspect_err(|err| {
|
||||
error!(
|
||||
@@ -97,8 +98,8 @@ pub(crate) async fn init_startup_storage_foundation(server_address: &str, volume
|
||||
);
|
||||
})
|
||||
.map_err(Error::other)?;
|
||||
prewarm_local_disk_id_map().await;
|
||||
init_lock_clients(endpoint_pools.clone());
|
||||
ecstore_storage::prewarm_local_disk_id_map().await;
|
||||
ecstore_storage::init_lock_clients(endpoint_pools.clone());
|
||||
|
||||
log_storage_pool_layout(&endpoint_pools);
|
||||
|
||||
@@ -114,13 +115,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}")))?;
|
||||
|
||||
set_global_endpoints(endpoint_pools.as_ref().clone());
|
||||
update_erasure_type(setup_type).await;
|
||||
ecstore_global::set_global_endpoints(endpoint_pools.as_ref().clone());
|
||||
ecstore_global::update_erasure_type(setup_type).await;
|
||||
|
||||
init_local_disks(endpoint_pools.clone())
|
||||
ecstore_storage::init_local_disks(endpoint_pools.clone())
|
||||
.await
|
||||
.map_err(|err| Error::other(format!("local disks: {err}")))?;
|
||||
init_lock_clients(endpoint_pools.clone());
|
||||
ecstore_storage::init_lock_clients(endpoint_pools.clone());
|
||||
|
||||
Ok(endpoint_pools)
|
||||
}
|
||||
@@ -158,7 +159,7 @@ pub(crate) async fn init_startup_storage_runtime(
|
||||
|
||||
init_startup_storage_global_config(store.clone()).await?;
|
||||
readiness.mark_stage(SystemStage::StorageReady);
|
||||
init_background_replication(store.clone()).await;
|
||||
ecstore_replication::init_background_replication(store.clone()).await;
|
||||
|
||||
Ok(StartupStorageRuntime {
|
||||
store,
|
||||
@@ -189,17 +190,17 @@ pub(crate) async fn init_embedded_startup_storage_runtime(
|
||||
|
||||
init_embedded_startup_storage_global_config(store.clone()).await?;
|
||||
readiness.mark_stage(SystemStage::StorageReady);
|
||||
init_background_replication(store.clone()).await;
|
||||
ecstore_replication::init_background_replication(store.clone()).await;
|
||||
|
||||
Ok(StartupStorageRuntime { store, shutdown_token })
|
||||
}
|
||||
|
||||
async fn init_startup_storage_global_config(store: Arc<ECStore>) -> Result<()> {
|
||||
init_ecstore_config();
|
||||
try_migrate_server_config(store.clone()).await;
|
||||
ecstore_config::init();
|
||||
ecstore_config::try_migrate_server_config(store.clone()).await;
|
||||
|
||||
let mut retry_count = 0;
|
||||
while let Err(e) = init_global_config_sys(store.clone()).await {
|
||||
while let Err(e) = ecstore_config::init_global_config_sys(store.clone()).await {
|
||||
let next_retry_count = retry_count + 1;
|
||||
error!(
|
||||
target: "rustfs::main::run",
|
||||
@@ -224,11 +225,11 @@ async fn init_startup_storage_global_config(store: Arc<ECStore>) -> Result<()> {
|
||||
}
|
||||
|
||||
async fn init_embedded_startup_storage_global_config(store: Arc<ECStore>) -> Result<()> {
|
||||
init_ecstore_config();
|
||||
try_migrate_server_config(store.clone()).await;
|
||||
ecstore_config::init();
|
||||
ecstore_config::try_migrate_server_config(store.clone()).await;
|
||||
|
||||
let mut retry = 0;
|
||||
while let Err(err) = init_global_config_sys(store.clone()).await {
|
||||
while let Err(err) = ecstore_config::init_global_config_sys(store.clone()).await {
|
||||
retry += 1;
|
||||
if retry > GLOBAL_CONFIG_INIT_MAX_RETRIES {
|
||||
return Err(Error::other(format!(
|
||||
|
||||
@@ -1,105 +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::bucket as ecstore_bucket;
|
||||
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::layout as ecstore_layout;
|
||||
use rustfs_ecstore::api::notification as ecstore_notification;
|
||||
use rustfs_ecstore::api::storage as ecstore_storage;
|
||||
|
||||
pub(crate) type ECStore = ecstore_storage::ECStore;
|
||||
#[cfg(test)]
|
||||
pub(crate) type EcstoreError = ecstore_error::Error;
|
||||
pub(crate) type EcstoreResult<T> = ecstore_error::Result<T>;
|
||||
pub(crate) type EndpointServerPools = ecstore_layout::EndpointServerPools;
|
||||
|
||||
pub(crate) async fn get_notification_config(bucket: &str) -> ecstore_error::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) 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 get_global_replication_pool() -> Option<Arc<ecstore_bucket::replication::DynReplicationPool>> {
|
||||
ecstore_bucket::replication::get_global_replication_pool()
|
||||
}
|
||||
|
||||
pub(crate) async fn init_background_replication(storage: Arc<ECStore>) {
|
||||
ecstore_bucket::replication::init_background_replication(storage).await;
|
||||
}
|
||||
|
||||
pub(crate) fn init_ecstore_config() {
|
||||
ecstore_config::init();
|
||||
}
|
||||
|
||||
pub(crate) async fn init_global_config_sys(api: Arc<ECStore>) -> EcstoreResult<()> {
|
||||
ecstore_config::init_global_config_sys(api).await
|
||||
}
|
||||
|
||||
pub(crate) async fn try_migrate_server_config(api: Arc<ECStore>) {
|
||||
ecstore_config::try_migrate_server_config(api).await;
|
||||
}
|
||||
|
||||
pub(crate) fn get_global_region() -> Option<s3s::region::Region> {
|
||||
ecstore_global::get_global_region()
|
||||
}
|
||||
|
||||
pub(crate) fn set_global_endpoints(eps: Vec<ecstore_layout::PoolEndpoints>) {
|
||||
ecstore_global::set_global_endpoints(eps);
|
||||
}
|
||||
|
||||
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) fn shutdown_background_services() {
|
||||
ecstore_global::shutdown_background_services();
|
||||
}
|
||||
|
||||
pub(crate) async fn update_erasure_type(setup_type: ecstore_layout::SetupType) {
|
||||
ecstore_global::update_erasure_type(setup_type).await
|
||||
}
|
||||
|
||||
pub(crate) async fn new_global_notification_sys(eps: EndpointServerPools) -> EcstoreResult<()> {
|
||||
ecstore_notification::new_global_notification_sys(eps).await
|
||||
}
|
||||
|
||||
pub(crate) async fn init_local_disks(endpoint_pools: EndpointServerPools) -> EcstoreResult<()> {
|
||||
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 prewarm_local_disk_id_map() {
|
||||
ecstore_storage::prewarm_local_disk_id_map().await;
|
||||
}
|
||||
+31
-24
@@ -27,11 +27,6 @@ use std::{
|
||||
time::{Duration as StdDuration, Instant},
|
||||
};
|
||||
|
||||
use super::table_catalog_storage_compat::{
|
||||
BUCKET_TABLE_CATALOG_META_PREFIX, BUCKET_TABLE_CATALOG_TABLE_BUCKETS_PREFIX, BUCKET_TABLE_CONFIG,
|
||||
BUCKET_TABLE_RESERVED_PREFIX, EcstoreError, RUSTFS_META_BUCKET, StorageError, get_bucket_metadata, get_lock_acquire_timeout,
|
||||
table_catalog_path_hash,
|
||||
};
|
||||
use bytes::Bytes;
|
||||
use datafusion::{
|
||||
arrow::datatypes::SchemaRef,
|
||||
@@ -39,6 +34,10 @@ use datafusion::{
|
||||
};
|
||||
use http::HeaderMap;
|
||||
use metrics::{counter, histogram};
|
||||
use rustfs_ecstore::api::{
|
||||
bucket::{metadata as ecstore_metadata, metadata_sys as ecstore_metadata_sys},
|
||||
disk as ecstore_disk, error as ecstore_error, set_disk as ecstore_set_disk,
|
||||
};
|
||||
use rustfs_filemeta::FileInfo;
|
||||
use rustfs_storage_api::{
|
||||
HTTPPreconditions, HTTPRangeSpec, ListObjectVersionsInfo as StorageListObjectVersionsInfo,
|
||||
@@ -56,7 +55,8 @@ use crate::storage::{
|
||||
StorageObjectOptions as ObjectOptions, StorageObjectToDelete as ObjectToDelete, StoragePutObjReader as PutObjReader,
|
||||
};
|
||||
|
||||
pub(crate) const TABLE_BUCKET_MARKER_CONFIG: &str = BUCKET_TABLE_CONFIG;
|
||||
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 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 = BUCKET_TABLE_RESERVED_PREFIX;
|
||||
pub const TABLE_RESERVED_PREFIX: &str = ecstore_metadata::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 = BUCKET_TABLE_CATALOG_META_PREFIX;
|
||||
const TABLE_BUCKET_ROOT: &str = BUCKET_TABLE_CATALOG_TABLE_BUCKETS_PREFIX;
|
||||
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 COMMIT_LOG_ROOT: &str = "commits";
|
||||
const COMMIT_IDEMPOTENCY_ROOT: &str = "commit-idempotency";
|
||||
const EXTERNAL_CATALOG_ROOT: &str = "external-catalog";
|
||||
@@ -115,8 +115,10 @@ 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<
|
||||
@@ -1461,7 +1463,7 @@ impl TableCatalogObjectPaths {
|
||||
"{}{}/{MAINTENANCE_ROOT}/{}/{MAINTENANCE_CONFIG_FILE}",
|
||||
self.table_entries_prefix(table_bucket, namespace),
|
||||
table.as_str(),
|
||||
table_catalog_path_hash(table_id)
|
||||
ecstore_metadata::table_catalog_path_hash(table_id)
|
||||
)
|
||||
}
|
||||
|
||||
@@ -1477,8 +1479,8 @@ impl TableCatalogObjectPaths {
|
||||
"{}{}/{MAINTENANCE_ROOT}/{}/{MAINTENANCE_JOB_ROOT}/{}.json",
|
||||
self.table_entries_prefix(table_bucket, namespace),
|
||||
table.as_str(),
|
||||
table_catalog_path_hash(table_id),
|
||||
table_catalog_path_hash(job_id)
|
||||
ecstore_metadata::table_catalog_path_hash(table_id),
|
||||
ecstore_metadata::table_catalog_path_hash(job_id)
|
||||
)
|
||||
}
|
||||
|
||||
@@ -1493,7 +1495,7 @@ impl TableCatalogObjectPaths {
|
||||
"{}{}/{MAINTENANCE_ROOT}/{}/{MAINTENANCE_LATEST_JOB_FILE}",
|
||||
self.table_entries_prefix(table_bucket, namespace),
|
||||
table.as_str(),
|
||||
table_catalog_path_hash(table_id)
|
||||
ecstore_metadata::table_catalog_path_hash(table_id)
|
||||
)
|
||||
}
|
||||
|
||||
@@ -1508,7 +1510,7 @@ impl TableCatalogObjectPaths {
|
||||
"{}{}/{MAINTENANCE_ROOT}/{}/{MAINTENANCE_CURRENT_JOB_FILE}",
|
||||
self.table_entries_prefix(table_bucket, namespace),
|
||||
table.as_str(),
|
||||
table_catalog_path_hash(table_id)
|
||||
ecstore_metadata::table_catalog_path_hash(table_id)
|
||||
)
|
||||
}
|
||||
|
||||
@@ -1517,8 +1519,8 @@ impl TableCatalogObjectPaths {
|
||||
"{}{}/{}/{}.json",
|
||||
self.table_bucket_root_prefix(table_bucket),
|
||||
COMMIT_LOG_ROOT,
|
||||
table_catalog_path_hash(table_id),
|
||||
table_catalog_path_hash(commit_id)
|
||||
ecstore_metadata::table_catalog_path_hash(table_id),
|
||||
ecstore_metadata::table_catalog_path_hash(commit_id)
|
||||
)
|
||||
}
|
||||
|
||||
@@ -1527,7 +1529,7 @@ impl TableCatalogObjectPaths {
|
||||
"{}{}/{}/",
|
||||
self.table_bucket_root_prefix(table_bucket),
|
||||
COMMIT_LOG_ROOT,
|
||||
table_catalog_path_hash(table_id)
|
||||
ecstore_metadata::table_catalog_path_hash(table_id)
|
||||
)
|
||||
}
|
||||
|
||||
@@ -1536,8 +1538,8 @@ impl TableCatalogObjectPaths {
|
||||
"{}{}/{}/{}.json",
|
||||
self.table_bucket_root_prefix(table_bucket),
|
||||
COMMIT_IDEMPOTENCY_ROOT,
|
||||
table_catalog_path_hash(table_id),
|
||||
table_catalog_path_hash(idempotency_key)
|
||||
ecstore_metadata::table_catalog_path_hash(table_id),
|
||||
ecstore_metadata::table_catalog_path_hash(idempotency_key)
|
||||
)
|
||||
}
|
||||
|
||||
@@ -1546,12 +1548,17 @@ impl TableCatalogObjectPaths {
|
||||
"{}{}/{}/",
|
||||
self.table_bucket_root_prefix(table_bucket),
|
||||
COMMIT_IDEMPOTENCY_ROOT,
|
||||
table_catalog_path_hash(table_id)
|
||||
ecstore_metadata::table_catalog_path_hash(table_id)
|
||||
)
|
||||
}
|
||||
|
||||
fn table_bucket_root_prefix(&self, table_bucket: &str) -> String {
|
||||
format!("{}/{}/{}/", self.catalog_root, TABLE_BUCKET_ROOT, table_catalog_path_hash(table_bucket))
|
||||
format!(
|
||||
"{}/{}/{}/",
|
||||
self.catalog_root,
|
||||
TABLE_BUCKET_ROOT,
|
||||
ecstore_metadata::table_catalog_path_hash(table_bucket)
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4031,7 +4038,7 @@ where
|
||||
.await
|
||||
.map_err(|err| storage_error_to_catalog("create catalog table lock", err))?;
|
||||
let guard = lock
|
||||
.get_write_lock(get_lock_acquire_timeout())
|
||||
.get_write_lock(ecstore_set_disk::get_lock_acquire_timeout())
|
||||
.await
|
||||
.map_err(|err| TableCatalogStoreError::Internal(format!("failed to acquire catalog table lock: {err}")))?;
|
||||
Ok(Box::new(guard))
|
||||
@@ -7002,7 +7009,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 = get_bucket_metadata(bucket)
|
||||
let table_bucket_enabled = ecstore_metadata_sys::get(bucket)
|
||||
.await
|
||||
.is_ok_and(|metadata| metadata.table_bucket_enabled());
|
||||
|
||||
@@ -7410,7 +7417,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/{}/", table_catalog_path_hash(bucket));
|
||||
let bucket_root = format!("s3tables/catalog/table-buckets/{}/", ecstore_metadata::table_catalog_path_hash(bucket));
|
||||
|
||||
assert_eq!(paths.table_bucket_entry_path(bucket), format!("{bucket_root}table-bucket.json"));
|
||||
assert_eq!(
|
||||
|
||||
@@ -1,41 +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, time::Duration};
|
||||
|
||||
use rustfs_ecstore::api::bucket as ecstore_bucket;
|
||||
use rustfs_ecstore::api::disk as ecstore_disk;
|
||||
use rustfs_ecstore::api::error as ecstore_error;
|
||||
use rustfs_ecstore::api::set_disk as ecstore_set_disk;
|
||||
|
||||
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 RUSTFS_META_BUCKET: &str = ecstore_disk::RUSTFS_META_BUCKET;
|
||||
pub(crate) type EcstoreError = ecstore_error::Error;
|
||||
pub(crate) type StorageError = ecstore_error::StorageError;
|
||||
|
||||
pub(crate) fn table_catalog_path_hash(value: &str) -> String {
|
||||
ecstore_bucket::metadata::table_catalog_path_hash(value)
|
||||
}
|
||||
|
||||
pub(crate) async fn get_bucket_metadata(bucket: &str) -> ecstore_error::Result<Arc<ecstore_bucket::metadata::BucketMetadata>> {
|
||||
ecstore_bucket::metadata_sys::get(bucket).await
|
||||
}
|
||||
|
||||
pub(crate) fn get_lock_acquire_timeout() -> Duration {
|
||||
ecstore_set_disk::get_lock_acquire_timeout()
|
||||
}
|
||||
@@ -16,10 +16,8 @@ use rustfs_concurrency::{
|
||||
AdmissionState, WorkloadAdmissionRegistrySnapshot, WorkloadAdmissionSnapshot, WorkloadAdmissionSnapshotProvider,
|
||||
WorkloadClass,
|
||||
};
|
||||
use rustfs_ecstore::api::bucket::{metadata_sys, replication};
|
||||
|
||||
use super::workload_admission_storage_compat::{
|
||||
get_global_bucket_metadata_sys, get_global_replication_pool, replication_queue_current_count,
|
||||
};
|
||||
use crate::storage::concurrency::get_concurrency_manager;
|
||||
|
||||
const BUCKET_METADATA_RUNTIME_NOT_INITIALIZED: &str = "bucket metadata runtime not initialized";
|
||||
@@ -77,7 +75,7 @@ pub fn foreground_read_workload_admission_snapshot() -> WorkloadAdmissionSnapsho
|
||||
}
|
||||
|
||||
pub fn metadata_workload_admission_snapshot() -> WorkloadAdmissionSnapshot {
|
||||
metadata_workload_admission_snapshot_from_initialized(get_global_bucket_metadata_sys().is_some())
|
||||
metadata_workload_admission_snapshot_from_initialized(metadata_sys::get_global_bucket_metadata_sys().is_some())
|
||||
}
|
||||
|
||||
fn metadata_workload_admission_snapshot_from_initialized(runtime_initialized: bool) -> WorkloadAdmissionSnapshot {
|
||||
@@ -153,7 +151,7 @@ fn repair_workload_admission_snapshot_from_counts(
|
||||
}
|
||||
|
||||
pub fn replication_workload_admission_snapshot() -> WorkloadAdmissionSnapshot {
|
||||
let Some(pool) = get_global_replication_pool() else {
|
||||
let Some(pool) = replication::get_global_replication_pool() else {
|
||||
return replication_workload_admission_snapshot_from_counts(false, None, None);
|
||||
};
|
||||
|
||||
@@ -200,6 +198,16 @@ 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::*;
|
||||
|
||||
@@ -1,36 +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::bucket as ecstore_bucket;
|
||||
|
||||
pub(crate) fn get_global_bucket_metadata_sys() -> Option<Arc<tokio::sync::RwLock<ecstore_bucket::metadata_sys::BucketMetadataSys>>>
|
||||
{
|
||||
ecstore_bucket::metadata_sys::get_global_bucket_metadata_sys()
|
||||
}
|
||||
|
||||
pub(crate) fn get_global_replication_pool() -> Option<Arc<ecstore_bucket::replication::DynReplicationPool>> {
|
||||
ecstore_bucket::replication::get_global_replication_pool()
|
||||
}
|
||||
|
||||
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())
|
||||
})
|
||||
}
|
||||
Reference in New Issue
Block a user