refactor: clean external operation consumers (#3565)

This commit is contained in:
安正超
2026-06-18 12:39:41 +08:00
committed by GitHub
parent d2a135b397
commit 99941f7e7c
8 changed files with 193 additions and 55 deletions
+32 -5
View File
@@ -33,8 +33,9 @@ use rustfs_ecstore::{
config::{com::save_config, storageclass},
disk::{BUCKET_META_PREFIX, RUSTFS_META_BUCKET},
error::{Error, Result as StorageResult, StorageError},
store_api::{ObjectIO, ObjectInfo, ObjectOptions},
store_api::{GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader},
};
use rustfs_storage_api::{HTTPRangeSpec, ObjectIO};
use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf};
use tokio::time::{Duration, Instant, sleep, timeout};
use tracing::warn;
@@ -60,6 +61,32 @@ const EVENT_SCANNER_CACHE_LOAD_STATE: &str = "scanner_cache_load_state";
const EVENT_SCANNER_CACHE_SAVE_STATE: &str = "scanner_cache_save_state";
static CACHE_SAVE_METRICS_ONCE: Once = Once::new();
pub trait ScannerObjectIO:
ObjectIO<
Error = Error,
RangeSpec = HTTPRangeSpec,
HeaderMap = HeaderMap,
ObjectOptions = ObjectOptions,
ObjectInfo = ObjectInfo,
GetObjectReader = GetObjectReader,
PutObjectReader = PutObjReader,
>
{
}
impl<T> ScannerObjectIO for T where
T: ObjectIO<
Error = Error,
RangeSpec = HTTPRangeSpec,
HeaderMap = HeaderMap,
ObjectOptions = ObjectOptions,
ObjectInfo = ObjectInfo,
GetObjectReader = GetObjectReader,
PutObjectReader = PutObjReader,
>
{
}
pub const DATA_USAGE_SCAN_CHECKPOINT_VERSION: u16 = 1;
// Data usage paths (computed at runtime)
@@ -616,7 +643,7 @@ impl DataUsageCache {
/// Only backend errors are returned as errors.
/// The loader is optimistic and has no locking, but tries 5 times before giving up.
/// If the object is not found, a nil error with empty data usage cache is returned.
pub async fn load<S: ObjectIO>(&mut self, store: Arc<S>, name: &str) -> StorageResult<()> {
pub async fn load<S: ScannerObjectIO>(&mut self, store: Arc<S>, name: &str) -> StorageResult<()> {
// By default, empty data usage cache
*self = DataUsageCache::default();
@@ -671,7 +698,7 @@ impl DataUsageCache {
}
// Inner load function that attempts to load from a specific path
// Returns (should_retry, cache_option, error_option)
async fn try_load_inner<S: ObjectIO>(
async fn try_load_inner<S: ScannerObjectIO>(
store: Arc<S>,
load_name: &str,
timeout_duration: Duration,
@@ -866,7 +893,7 @@ impl DataUsageCache {
Err(last_err.unwrap_or_else(|| StorageError::other("Failed to save data usage cache".to_string())))
}
async fn save_path_with_retry<S: ObjectIO>(
async fn save_path_with_retry<S: ScannerObjectIO>(
store: Arc<S>,
path: &str,
buf: &[u8],
@@ -889,7 +916,7 @@ impl DataUsageCache {
.await
}
pub async fn save<S: ObjectIO>(&self, store: Arc<S>, name: &str) -> StorageResult<()> {
pub async fn save<S: ScannerObjectIO>(&self, store: Arc<S>, name: &str) -> StorageResult<()> {
let mut buf = Vec::new();
self.serialize(&mut rmp_serde::Serializer::new(&mut buf))?;
let timeout_duration = Self::cache_save_timeout();
+4 -3
View File
@@ -14,7 +14,9 @@
use std::sync::Arc;
use crate::data_usage_define::{BACKGROUND_HEAL_INFO_PATH, DATA_USAGE_BLOOM_NAME_PATH, DATA_USAGE_OBJ_NAME_PATH};
use crate::data_usage_define::{
BACKGROUND_HEAL_INFO_PATH, DATA_USAGE_BLOOM_NAME_PATH, DATA_USAGE_OBJ_NAME_PATH, ScannerObjectIO,
};
use crate::runtime_config::{
current_scanner_runtime_config, lookup_scanner_runtime_config, refresh_scanner_runtime_config_from_global,
scanner_bitrot_cycle, scanner_cycle_interval, scanner_start_delay, set_scanner_default_cycle_secs,
@@ -45,7 +47,6 @@ use rustfs_ecstore::disk::RUSTFS_META_BUCKET;
use rustfs_ecstore::error::Error as EcstoreError;
use rustfs_ecstore::global::is_erasure_sd;
use rustfs_ecstore::store::ECStore;
use rustfs_ecstore::store_api::ObjectIO;
use rustfs_storage_api::{BucketOperations, BucketOptions, NamespaceLocking as _};
use serde::{Deserialize, Serialize};
use tokio::sync::mpsc;
@@ -976,7 +977,7 @@ impl Drop for ScannerScanModeGuard {
#[instrument(skip(ctx, storeapi))]
pub async fn store_data_usage_in_backend(
ctx: CancellationToken,
storeapi: Arc<impl ObjectIO>,
storeapi: Arc<impl ScannerObjectIO>,
mut receiver: mpsc::Receiver<DataUsageInfo>,
) {
let mut attempts = 1u32;
+3 -3
View File
@@ -17,7 +17,7 @@ use crate::scanner_folder::{ScannerItem, scan_data_folder};
use crate::sleeper::SCANNER_SLEEPER;
use crate::{
DATA_USAGE_CACHE_NAME, DATA_USAGE_ROOT, DataUsageCache, DataUsageCacheInfo, DataUsageEntry, DataUsageEntryInfo,
DataUsageInfo, ScannerError, SizeSummary, TierStats,
DataUsageInfo, ScannerError, ScannerObjectIO, SizeSummary, TierStats,
};
use futures::future::join_all;
use metrics::counter;
@@ -41,7 +41,7 @@ use rustfs_ecstore::error::{Error, StorageError};
use rustfs_ecstore::global::GLOBAL_TierConfigMgr;
use rustfs_ecstore::resolve_object_store_handle;
use rustfs_ecstore::set_disk::SetDisks;
use rustfs_ecstore::store_api::{ObjectIO, ObjectInfo};
use rustfs_ecstore::store_api::ObjectInfo;
use rustfs_ecstore::{error::Result, store::ECStore};
use rustfs_filemeta::FileMeta;
use rustfs_storage_api::{BucketInfo, BucketOperations, BucketOptions, DiskSetSelector, StorageAdminApi};
@@ -407,7 +407,7 @@ async fn send_cache_root_entry_info(
bucket_result_tx.lock().await.send(cache_root_entry_info(cache)).await
}
async fn persist_and_publish_cache_snapshot<S: ObjectIO>(
async fn persist_and_publish_cache_snapshot<S: ScannerObjectIO>(
store: Arc<S>,
updates: &mpsc::Sender<DataUsageCache>,
cache_snapshot: DataUsageCache,