mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-13 16:46:55 +00:00
refactor: route scanner heal storage boundaries (#3888)
This commit is contained in:
Zhengchao An
+20
-55
@@ -22,61 +22,25 @@
|
||||
|
||||
use http::HeaderMap;
|
||||
use rustfs_config::server_config::{Config as ServerConfig, get_global_server_config as config_get_global_server_config};
|
||||
use rustfs_ecstore::api::bucket::bucket_target_sys::BucketTargetSys as EcstoreBucketTargetSys;
|
||||
use rustfs_ecstore::api::bucket::lifecycle::bucket_lifecycle_audit::LcEventSrc as EcstoreLcEventSrc;
|
||||
use rustfs_ecstore::api::bucket::lifecycle::bucket_lifecycle_ops::{
|
||||
apply_expiry_rule as ecstore_apply_expiry_rule, apply_transition_rule as ecstore_apply_transition_rule,
|
||||
get_global_expiry_state as ecstore_get_global_expiry_state,
|
||||
};
|
||||
use rustfs_ecstore::api::bucket::lifecycle::evaluator::Evaluator as EcstoreEvaluator;
|
||||
use rustfs_ecstore::api::bucket::lifecycle::lifecycle::{
|
||||
Event as EcstoreEvent, Lifecycle as EcstoreLifecycle, ObjectOpts as EcstoreObjectOpts,
|
||||
TRANSITION_COMPLETE as ECSTORE_TRANSITION_COMPLETE,
|
||||
};
|
||||
use rustfs_ecstore::api::bucket::metadata_sys::{
|
||||
get_lifecycle_config as ecstore_get_lifecycle_config, get_object_lock_config as ecstore_get_object_lock_config,
|
||||
get_replication_config as ecstore_get_replication_config,
|
||||
};
|
||||
use rustfs_ecstore::api::bucket::replication::{
|
||||
ReplicationConfig as EcstoreReplicationConfig, ReplicationConfigurationExt as EcstoreReplicationConfigurationExt,
|
||||
ReplicationHealQueueResult as EcstoreReplicationHealQueueResult,
|
||||
ReplicationQueueAdmission as EcstoreReplicationQueueAdmission,
|
||||
queue_replication_heal_internal as ecstore_queue_replication_heal_internal,
|
||||
};
|
||||
use rustfs_ecstore::api::bucket::versioning::VersioningApi as EcstoreVersioningApi;
|
||||
use rustfs_ecstore::api::bucket::versioning_sys::BucketVersioningSys as EcstoreBucketVersioningSys;
|
||||
use rustfs_ecstore::api::cache::{ListPathRawOptions as EcstoreListPathRawOptions, list_path_raw as ecstore_list_path_raw};
|
||||
use rustfs_ecstore::api::capacity::{
|
||||
is_reserved_or_invalid_bucket as ecstore_is_reserved_or_invalid_bucket, path2_bucket_object as ecstore_path2_bucket_object,
|
||||
path2_bucket_object_with_base_path as ecstore_path2_bucket_object_with_base_path,
|
||||
};
|
||||
use rustfs_ecstore::api::config::com::{read_config as ecstore_read_config, save_config as ecstore_save_config};
|
||||
#[cfg(test)]
|
||||
use rustfs_ecstore::api::config::init as ecstore_config_init;
|
||||
use rustfs_ecstore::api::config::storageclass::{RRS as ECSTORE_STORAGECLASS_RRS, STANDARD as ECSTORE_STORAGECLASS_STANDARD};
|
||||
use rustfs_ecstore::api::data_usage::replace_bucket_usage_memory_from_info as ecstore_replace_bucket_usage_memory_from_info;
|
||||
#[cfg(test)]
|
||||
use rustfs_ecstore::api::disk::endpoint::Endpoint as EcstoreEndpoint;
|
||||
use rustfs_ecstore::api::disk::error::{DiskError as EcstoreDiskError, Result as EcstoreDiskResult};
|
||||
use rustfs_ecstore::api::disk::{
|
||||
BUCKET_META_PREFIX as ECSTORE_BUCKET_META_PREFIX, Bytes as EcstoreDiskBytes, Disk as EcstoreDisk, DiskAPI as EcstoreDiskAPI,
|
||||
DiskInfo as EcstoreDiskInfo, DiskInfoOptions as EcstoreDiskInfoOptions, DiskLocation as EcstoreDiskLocation,
|
||||
RUSTFS_META_BUCKET as ECSTORE_RUSTFS_META_BUCKET, STORAGE_FORMAT_FILE as ECSTORE_STORAGE_FORMAT_FILE,
|
||||
ScanGuard as EcstoreScanGuard,
|
||||
};
|
||||
#[cfg(test)]
|
||||
use rustfs_ecstore::api::disk::{DiskOption as EcstoreDiskOption, DiskStore as EcstoreDiskStore, new_disk as ecstore_new_disk};
|
||||
use rustfs_ecstore::api::error::{Error as EcstoreErrorType, Result as EcstoreResultType, StorageError as EcstoreStorageError};
|
||||
use rustfs_ecstore::api::global::{
|
||||
get_global_tier_config_mgr as ecstore_get_global_tier_config_mgr, is_erasure as ecstore_is_erasure,
|
||||
is_erasure_sd as ecstore_is_erasure_sd, resolve_object_store_handle as ecstore_resolve_object_store_handle,
|
||||
};
|
||||
use rustfs_ecstore::api::set_disk::SetDisks as EcstoreSetDisks;
|
||||
use rustfs_ecstore::api::storage::ECStore as EcstoreStore;
|
||||
use rustfs_ecstore::api::tier::tier_config::TierConfig as EcstoreTierConfig;
|
||||
use rustfs_storage_api::{HTTPRangeSpec, ObjectIO, ObjectToDelete};
|
||||
use std::path::PathBuf;
|
||||
use std::sync::Arc;
|
||||
use storage_api::{
|
||||
ECSTORE_BUCKET_META_PREFIX, ECSTORE_RUSTFS_META_BUCKET, ECSTORE_STORAGE_FORMAT_FILE, ECSTORE_STORAGECLASS_RRS,
|
||||
ECSTORE_STORAGECLASS_STANDARD, ECSTORE_TRANSITION_COMPLETE, EcstoreBucketTargetSys, EcstoreBucketVersioningSys, EcstoreDisk,
|
||||
EcstoreDiskAPI, EcstoreDiskBytes, EcstoreDiskError, EcstoreDiskInfo, EcstoreDiskInfoOptions, EcstoreDiskLocation,
|
||||
EcstoreDiskResult, EcstoreErrorType, EcstoreEvaluator, EcstoreEvent, EcstoreLcEventSrc, EcstoreLifecycle,
|
||||
EcstoreListPathRawOptions, EcstoreObjectOpts, EcstoreReplicationConfig, EcstoreReplicationConfigurationExt,
|
||||
EcstoreReplicationHealQueueResult, EcstoreReplicationQueueAdmission, EcstoreResultType, EcstoreScanGuard, EcstoreSetDisks,
|
||||
EcstoreStorageError, EcstoreStore, EcstoreTierConfig, EcstoreVersioningApi, HTTPRangeSpec, ObjectIO, ObjectOperations,
|
||||
ObjectToDelete, ecstore_apply_expiry_rule, ecstore_apply_transition_rule, ecstore_get_global_expiry_state,
|
||||
ecstore_get_global_tier_config_mgr, ecstore_get_lifecycle_config, ecstore_get_object_lock_config,
|
||||
ecstore_get_replication_config, ecstore_is_erasure, ecstore_is_erasure_sd, ecstore_is_reserved_or_invalid_bucket,
|
||||
ecstore_list_path_raw, ecstore_path2_bucket_object, ecstore_path2_bucket_object_with_base_path,
|
||||
ecstore_queue_replication_heal_internal, ecstore_read_config, ecstore_replace_bucket_usage_memory_from_info,
|
||||
ecstore_resolve_object_store_handle, ecstore_save_config,
|
||||
};
|
||||
#[cfg(test)]
|
||||
use storage_api::{EcstoreDiskOption, EcstoreDiskStore, EcstoreEndpoint, ecstore_config_init, ecstore_new_disk};
|
||||
use tokio_util::sync::CancellationToken;
|
||||
|
||||
pub mod data_usage_define;
|
||||
@@ -87,6 +51,7 @@ pub mod scanner_budget;
|
||||
pub mod scanner_folder;
|
||||
pub mod scanner_io;
|
||||
pub mod sleeper;
|
||||
pub(crate) mod storage_api;
|
||||
|
||||
pub use data_usage_define::*;
|
||||
pub use error::ScannerError;
|
||||
@@ -151,8 +116,8 @@ pub(crate) type SetDisks = EcstoreSetDisks;
|
||||
pub(crate) type StorageError = EcstoreStorageError;
|
||||
|
||||
pub type ScannerGetObjectReader = <ECStore as ObjectIO>::GetObjectReader;
|
||||
pub type ScannerObjectInfo = <ECStore as rustfs_storage_api::ObjectOperations>::ObjectInfo;
|
||||
pub type ScannerObjectOptions = <ECStore as rustfs_storage_api::ObjectOperations>::ObjectOptions;
|
||||
pub type ScannerObjectInfo = <ECStore as ObjectOperations>::ObjectInfo;
|
||||
pub type ScannerObjectOptions = <ECStore as ObjectOperations>::ObjectOptions;
|
||||
pub type ScannerObjectToDelete = ObjectToDelete;
|
||||
pub type ScannerPutObjReader = <ECStore as ObjectIO>::PutObjectReader;
|
||||
|
||||
|
||||
@@ -38,13 +38,13 @@ use rustfs_config::{
|
||||
ENV_SCANNER_CYCLE_MAX_OBJECTS,
|
||||
};
|
||||
use rustfs_config::{ENV_SCANNER_CYCLE, ENV_SCANNER_SPEED, ENV_SCANNER_START_DELAY_SECS};
|
||||
use rustfs_storage_api::{BucketOperations, BucketOptions, NamespaceLocking as _};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use tokio::sync::mpsc;
|
||||
use tokio::time::{Duration, Instant};
|
||||
use tokio_util::sync::CancellationToken;
|
||||
use tracing::{debug, error, info, instrument, warn};
|
||||
|
||||
use crate::storage_api::{BucketOperations, BucketOptions, NamespaceLocking as _};
|
||||
use crate::{
|
||||
ECStore, EcstoreError, RUSTFS_META_BUCKET, ScannerLifecycleConfigExt as _, ScannerReplicationConfigExt as _,
|
||||
get_lifecycle_config, get_replication_config, read_config, replace_bucket_usage_memory_from_info, save_config,
|
||||
@@ -1148,9 +1148,9 @@ mod tests {
|
||||
}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl rustfs_storage_api::ObjectIO for MemoryConfigStore {
|
||||
impl crate::storage_api::ObjectIO for MemoryConfigStore {
|
||||
type Error = EcstoreError;
|
||||
type RangeSpec = rustfs_storage_api::HTTPRangeSpec;
|
||||
type RangeSpec = crate::storage_api::HTTPRangeSpec;
|
||||
type HeaderMap = http::HeaderMap;
|
||||
type ObjectOptions = ObjectOptions;
|
||||
type ObjectInfo = ObjectInfo;
|
||||
@@ -1161,7 +1161,7 @@ mod tests {
|
||||
&self,
|
||||
bucket: &str,
|
||||
object: &str,
|
||||
_range: Option<rustfs_storage_api::HTTPRangeSpec>,
|
||||
_range: Option<crate::storage_api::HTTPRangeSpec>,
|
||||
_h: http::HeaderMap,
|
||||
_opts: &ObjectOptions,
|
||||
) -> EcstoreResult<GetObjectReader> {
|
||||
|
||||
@@ -27,7 +27,6 @@ use rustfs_common::metrics::{Metric, Metrics, emit_scan_bucket_drive_complete, e
|
||||
#[cfg(test)]
|
||||
use rustfs_config::{ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS, ENV_SCANNER_MAX_CONCURRENT_SET_SCANS};
|
||||
use rustfs_filemeta::FileMeta;
|
||||
use rustfs_storage_api::{BucketInfo, BucketOperations, BucketOptions, DiskSetSelector, StorageAdminApi};
|
||||
use rustfs_utils::path::path_join_buf;
|
||||
use s3s::dto::{BucketLifecycleConfiguration, ReplicationConfiguration};
|
||||
use std::collections::HashMap;
|
||||
@@ -43,6 +42,7 @@ use tokio_util::sync::CancellationToken;
|
||||
use tracing::{debug, error, warn};
|
||||
|
||||
use crate::ScannerObjectInfo as ObjectInfo;
|
||||
use crate::storage_api::{BucketInfo, BucketOperations, BucketOptions, DiskSetSelector, StorageAdminApi};
|
||||
use crate::{
|
||||
BucketTargetSys, BucketVersioningSys, Disk, DiskError, ECStore, EcstoreError as Error, EcstoreResult as Result,
|
||||
ReplicationConfig, STORAGE_FORMAT_FILE, ScannerDiskExt as _, ScannerLifecycleConfigExt as _,
|
||||
|
||||
@@ -0,0 +1,78 @@
|
||||
// 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) use rustfs_ecstore::api::bucket::bucket_target_sys::BucketTargetSys as EcstoreBucketTargetSys;
|
||||
pub(crate) use rustfs_ecstore::api::bucket::lifecycle::bucket_lifecycle_audit::LcEventSrc as EcstoreLcEventSrc;
|
||||
pub(crate) use rustfs_ecstore::api::bucket::lifecycle::bucket_lifecycle_ops::{
|
||||
apply_expiry_rule as ecstore_apply_expiry_rule, apply_transition_rule as ecstore_apply_transition_rule,
|
||||
get_global_expiry_state as ecstore_get_global_expiry_state,
|
||||
};
|
||||
pub(crate) use rustfs_ecstore::api::bucket::lifecycle::evaluator::Evaluator as EcstoreEvaluator;
|
||||
pub(crate) use rustfs_ecstore::api::bucket::lifecycle::lifecycle::{
|
||||
Event as EcstoreEvent, Lifecycle as EcstoreLifecycle, ObjectOpts as EcstoreObjectOpts,
|
||||
TRANSITION_COMPLETE as ECSTORE_TRANSITION_COMPLETE,
|
||||
};
|
||||
pub(crate) use rustfs_ecstore::api::bucket::metadata_sys::{
|
||||
get_lifecycle_config as ecstore_get_lifecycle_config, get_object_lock_config as ecstore_get_object_lock_config,
|
||||
get_replication_config as ecstore_get_replication_config,
|
||||
};
|
||||
pub(crate) use rustfs_ecstore::api::bucket::replication::{
|
||||
ReplicationConfig as EcstoreReplicationConfig, ReplicationConfigurationExt as EcstoreReplicationConfigurationExt,
|
||||
ReplicationHealQueueResult as EcstoreReplicationHealQueueResult,
|
||||
ReplicationQueueAdmission as EcstoreReplicationQueueAdmission,
|
||||
queue_replication_heal_internal as ecstore_queue_replication_heal_internal,
|
||||
};
|
||||
pub(crate) use rustfs_ecstore::api::bucket::versioning::VersioningApi as EcstoreVersioningApi;
|
||||
pub(crate) use rustfs_ecstore::api::bucket::versioning_sys::BucketVersioningSys as EcstoreBucketVersioningSys;
|
||||
pub(crate) use rustfs_ecstore::api::cache::{
|
||||
ListPathRawOptions as EcstoreListPathRawOptions, list_path_raw as ecstore_list_path_raw,
|
||||
};
|
||||
pub(crate) use rustfs_ecstore::api::capacity::{
|
||||
is_reserved_or_invalid_bucket as ecstore_is_reserved_or_invalid_bucket, path2_bucket_object as ecstore_path2_bucket_object,
|
||||
path2_bucket_object_with_base_path as ecstore_path2_bucket_object_with_base_path,
|
||||
};
|
||||
pub(crate) use rustfs_ecstore::api::config::com::{read_config as ecstore_read_config, save_config as ecstore_save_config};
|
||||
#[cfg(test)]
|
||||
pub(crate) use rustfs_ecstore::api::config::init as ecstore_config_init;
|
||||
pub(crate) use rustfs_ecstore::api::config::storageclass::{
|
||||
RRS as ECSTORE_STORAGECLASS_RRS, STANDARD as ECSTORE_STORAGECLASS_STANDARD,
|
||||
};
|
||||
pub(crate) use rustfs_ecstore::api::data_usage::replace_bucket_usage_memory_from_info as ecstore_replace_bucket_usage_memory_from_info;
|
||||
#[cfg(test)]
|
||||
pub(crate) use rustfs_ecstore::api::disk::endpoint::Endpoint as EcstoreEndpoint;
|
||||
pub(crate) use rustfs_ecstore::api::disk::error::{DiskError as EcstoreDiskError, Result as EcstoreDiskResult};
|
||||
pub(crate) use rustfs_ecstore::api::disk::{
|
||||
BUCKET_META_PREFIX as ECSTORE_BUCKET_META_PREFIX, Bytes as EcstoreDiskBytes, Disk as EcstoreDisk, DiskAPI as EcstoreDiskAPI,
|
||||
DiskInfo as EcstoreDiskInfo, DiskInfoOptions as EcstoreDiskInfoOptions, DiskLocation as EcstoreDiskLocation,
|
||||
RUSTFS_META_BUCKET as ECSTORE_RUSTFS_META_BUCKET, STORAGE_FORMAT_FILE as ECSTORE_STORAGE_FORMAT_FILE,
|
||||
ScanGuard as EcstoreScanGuard,
|
||||
};
|
||||
#[cfg(test)]
|
||||
pub(crate) use rustfs_ecstore::api::disk::{
|
||||
DiskOption as EcstoreDiskOption, DiskStore as EcstoreDiskStore, new_disk as ecstore_new_disk,
|
||||
};
|
||||
pub(crate) use rustfs_ecstore::api::error::{
|
||||
Error as EcstoreErrorType, Result as EcstoreResultType, StorageError as EcstoreStorageError,
|
||||
};
|
||||
pub(crate) use rustfs_ecstore::api::global::{
|
||||
get_global_tier_config_mgr as ecstore_get_global_tier_config_mgr, is_erasure as ecstore_is_erasure,
|
||||
is_erasure_sd as ecstore_is_erasure_sd, resolve_object_store_handle as ecstore_resolve_object_store_handle,
|
||||
};
|
||||
pub(crate) use rustfs_ecstore::api::set_disk::SetDisks as EcstoreSetDisks;
|
||||
pub(crate) use rustfs_ecstore::api::storage::ECStore as EcstoreStore;
|
||||
pub(crate) use rustfs_ecstore::api::tier::tier_config::TierConfig as EcstoreTierConfig;
|
||||
pub(crate) use rustfs_storage_api::{
|
||||
BucketInfo, BucketOperations, BucketOptions, DiskSetSelector, HTTPRangeSpec, NamespaceLocking, ObjectIO, ObjectOperations,
|
||||
ObjectToDelete, StorageAdminApi,
|
||||
};
|
||||
Reference in New Issue
Block a user