refactor: wrap disk rpc compat trait methods (#3707)

This commit is contained in:
安正超
2026-06-22 06:31:24 +08:00
committed by GitHub
parent 7eb9a4759e
commit c13f42e9a0
18 changed files with 460 additions and 52 deletions
+5 -3
View File
@@ -77,12 +77,14 @@ pub mod disk {
pub use crate::disk::endpoint::Endpoint;
pub use crate::disk::error::DiskError;
pub use crate::disk::error_reduce::is_all_buckets_not_found;
pub use crate::disk::local::ScanGuard;
pub use crate::disk::{
BUCKET_META_PREFIX, DeleteOptions, Disk, DiskAPI, DiskInfoOptions, DiskOption, DiskStore, FileInfoVersions,
RUSTFS_META_BUCKET, ReadMultipleReq, ReadMultipleResp, ReadOptions, STORAGE_FORMAT_FILE, UpdateMetadataOpts, VolumeInfo,
WalkDirOptions, new_disk,
BUCKET_META_PREFIX, CheckPartsResp, DeleteOptions, Disk, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskOption,
DiskStore, FileInfoVersions, FileReader, FileWriter, RUSTFS_META_BUCKET, ReadMultipleReq, ReadMultipleResp, ReadOptions,
RenameDataResp, STORAGE_FORMAT_FILE, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, new_disk,
};
pub use crate::disk::{endpoint, error, error_reduce};
pub use bytes::Bytes;
}
pub mod error {
+1 -2
View File
@@ -20,7 +20,6 @@ use crate::heal::{
use crate::{Error, Result};
use metrics::{counter, gauge};
use rustfs_common::heal_channel::{HealAdmissionDropReason, HealAdmissionResult, HealRequestSource};
use rustfs_ecstore::api::disk::DiskAPI as _;
use rustfs_madmin::heal_commands::HealResultItem;
use std::{
collections::{BinaryHeap, HashMap},
@@ -34,7 +33,7 @@ use tokio::{
use tokio_util::sync::CancellationToken;
use tracing::{debug, error, info, warn};
use super::storage_compat::{DiskError, GLOBAL_LOCAL_DISK_MAP};
use super::storage_compat::{DiskError, GLOBAL_LOCAL_DISK_MAP, HealDiskExt as _};
const KEEP_HEAL_TASK_STATUS_DURATION: Duration = Duration::from_secs(10 * 60);
const LOG_COMPONENT_HEAL: &str = "heal";
+1 -2
View File
@@ -13,7 +13,6 @@
// limitations under the License.
use crate::{Error, Result};
use rustfs_ecstore::api::disk::DiskAPI as _;
use serde::{Deserialize, Serialize};
use std::path::Path;
use std::sync::Arc;
@@ -22,7 +21,7 @@ use tokio::sync::RwLock;
use tracing::{debug, warn};
use uuid::Uuid;
use super::storage_compat::{BUCKET_META_PREFIX, DiskError, DiskStore, RUSTFS_META_BUCKET};
use super::storage_compat::{BUCKET_META_PREFIX, DiskError, DiskStore, HealDiskExt as _, RUSTFS_META_BUCKET};
const LOG_COMPONENT_HEAL: &str = "heal";
const LOG_SUBSYSTEM_RESUME: &str = "resume";
+46
View File
@@ -22,6 +22,7 @@ pub(crate) const BUCKET_META_PREFIX: &str = ecstore_disk::BUCKET_META_PREFIX;
pub(crate) const RUSTFS_META_BUCKET: &str = ecstore_disk::RUSTFS_META_BUCKET;
pub(crate) type DiskError = ecstore_disk::error::DiskError;
pub(crate) type DiskResult<T> = ecstore_disk::error::Result<T>;
pub(crate) type DiskStore = ecstore_disk::DiskStore;
pub(crate) type ECStore = ecstore_storage::ECStore;
pub(crate) type EcstoreError = ecstore_error::Error;
@@ -47,6 +48,51 @@ pub(crate) async fn new_disk(ep: &Endpoint, opt: &DiskOption) -> ecstore_disk::e
ecstore_disk::new_disk(ep, opt).await
}
pub(crate) trait HealDiskExt {
fn endpoint(&self) -> Endpoint;
async fn get_disk_id(&self) -> DiskResult<Option<uuid::Uuid>>;
async fn read_all(&self, volume: &str, path: &str) -> DiskResult<ecstore_disk::Bytes>;
async fn write_all(&self, volume: &str, path: &str, data: ecstore_disk::Bytes) -> DiskResult<()>;
async fn delete(&self, volume: &str, path: &str, options: ecstore_disk::DeleteOptions) -> DiskResult<()>;
async fn list_dir(&self, origvolume: &str, volume: &str, dir_path: &str, count: i32) -> DiskResult<Vec<String>>;
#[cfg(test)]
async fn make_volume(&self, volume: &str) -> DiskResult<()>;
}
impl<T> HealDiskExt for T
where
T: ecstore_disk::DiskAPI,
{
fn endpoint(&self) -> Endpoint {
ecstore_disk::DiskAPI::endpoint(self)
}
async fn get_disk_id(&self) -> DiskResult<Option<uuid::Uuid>> {
ecstore_disk::DiskAPI::get_disk_id(self).await
}
async fn read_all(&self, volume: &str, path: &str) -> DiskResult<ecstore_disk::Bytes> {
ecstore_disk::DiskAPI::read_all(self, volume, path).await
}
async fn write_all(&self, volume: &str, path: &str, data: ecstore_disk::Bytes) -> DiskResult<()> {
ecstore_disk::DiskAPI::write_all(self, volume, path, data).await
}
async fn delete(&self, volume: &str, path: &str, options: ecstore_disk::DeleteOptions) -> DiskResult<()> {
ecstore_disk::DiskAPI::delete(self, volume, path, options).await
}
async fn list_dir(&self, origvolume: &str, volume: &str, dir_path: &str, count: i32) -> DiskResult<Vec<String>> {
ecstore_disk::DiskAPI::list_dir(self, origvolume, volume, dir_path, count).await
}
#[cfg(test)]
async fn make_volume(&self, volume: &str) -> DiskResult<()> {
ecstore_disk::DiskAPI::make_volume(self, volume).await
}
}
pub type HealObjectInfo = <ECStore as rustfs_storage_api::ObjectOperations>::ObjectInfo;
pub type HealObjectOptions = <ECStore as rustfs_storage_api::ObjectOperations>::ObjectOptions;
pub type HealPutObjReader = <ECStore as rustfs_storage_api::ObjectIO>::PutObjectReader;
+4 -5
View File
@@ -39,7 +39,6 @@ use rustfs_common::metrics::{
IlmAction, Metric, Metrics, ScannerReplicationRepairKind, ScannerSourceWorkUpdate, ScannerWorkSource, UpdateCurrentPathFn,
current_path_updater, global_metrics,
};
use rustfs_ecstore::api::disk::DiskAPI as _;
use rustfs_filemeta::{
MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams, ReplicateObjectInfo, ReplicationStatusType, ReplicationType,
};
@@ -53,10 +52,10 @@ use tracing::{debug, error, warn};
use crate::storage_compat::{
BucketVersioningSys, Disk, DiskError, DiskInfoOptions, Evaluator, Event, LcEventSrc, ListPathRawOptions, ObjectOpts,
ReplicationConfig, ReplicationQueueAdmission, ScannerLifecycleConfigExt as _, ScannerReplicationConfigExt as _,
ScannerVersioningConfigExt as _, StorageError, apply_expiry_rule, apply_transition_rule, enqueue_global_newer_noncurrent,
is_erasure, is_reserved_or_invalid_bucket, list_path_raw, path2_bucket_object, path2_bucket_object_with_base_path,
queue_replication_heal_internal,
ReplicationConfig, ReplicationQueueAdmission, ScannerDiskExt as _, ScannerLifecycleConfigExt as _,
ScannerReplicationConfigExt as _, ScannerVersioningConfigExt as _, StorageError, apply_expiry_rule, apply_transition_rule,
enqueue_global_newer_noncurrent, is_erasure, is_reserved_or_invalid_bucket, list_path_raw, path2_bucket_object,
path2_bucket_object_with_base_path, queue_replication_heal_internal,
};
use crate::{ScannerObjectInfo as ObjectInfo, ScannerObjectToDelete as ObjectToDelete};
+4 -4
View File
@@ -26,7 +26,6 @@ use rustfs_common::heal_channel::HealScanMode;
use rustfs_common::metrics::{Metric, Metrics, emit_scan_bucket_drive_complete, emit_scan_bucket_drive_partial, global_metrics};
#[cfg(test)]
use rustfs_config::{ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS, ENV_SCANNER_MAX_CONCURRENT_SET_SCANS};
use rustfs_ecstore::api::disk::DiskAPI as _;
use rustfs_filemeta::FileMeta;
use rustfs_storage_api::{BucketInfo, BucketOperations, BucketOptions, DiskSetSelector, StorageAdminApi};
use rustfs_utils::path::path_join_buf;
@@ -46,9 +45,10 @@ use tracing::{debug, error, warn};
use crate::ScannerObjectInfo as ObjectInfo;
use crate::storage_compat::{
BucketTargetSys, BucketVersioningSys, Disk, DiskError, ECStore, EcstoreError as Error, EcstoreResult as Result,
ReplicationConfig, STORAGE_FORMAT_FILE, ScannerLifecycleConfigExt as _, ScannerReplicationConfigExt as _,
ScannerVersioningConfigExt as _, SetDisks, StorageError, enqueue_global_free_version, get_lifecycle_config,
get_object_lock_config, get_replication_config, list_global_tiers, resolve_scanner_object_store_handle, storageclass,
ReplicationConfig, STORAGE_FORMAT_FILE, ScannerDiskExt as _, ScannerLifecycleConfigExt as _,
ScannerReplicationConfigExt as _, ScannerVersioningConfigExt as _, SetDisks, StorageError, enqueue_global_free_version,
get_lifecycle_config, get_object_lock_config, get_replication_config, list_global_tiers, resolve_scanner_object_store_handle,
storageclass,
};
pub(crate) const SCANNER_SKIP_FILE_ERROR: &str = "skip file";
+37
View File
@@ -19,6 +19,7 @@ use rustfs_ecstore::api::{
set_disk as ecstore_set_disk, storage as ecstore_storage, tier as ecstore_tier,
};
use rustfs_storage_api::{HTTPRangeSpec, ObjectIO, ObjectToDelete};
use std::path::PathBuf;
use std::sync::Arc;
use tokio_util::sync::CancellationToken;
@@ -30,7 +31,9 @@ pub(crate) const TRANSITION_COMPLETE: &str = ecstore_bucket::lifecycle::lifecycl
pub(crate) type Disk = ecstore_disk::Disk;
#[cfg(test)]
pub(crate) type DiskStore = ecstore_disk::DiskStore;
pub(crate) type DiskLocation = ecstore_disk::DiskLocation;
pub(crate) type DiskError = ecstore_disk::error::DiskError;
pub(crate) type DiskResult<T> = ecstore_disk::error::Result<T>;
pub(crate) type ECStore = ecstore_storage::ECStore;
pub(crate) type EcstoreError = ecstore_error::Error;
pub(crate) type EcstoreResult<T> = ecstore_error::Result<T>;
@@ -45,6 +48,7 @@ pub(crate) type ObjectOpts = ecstore_bucket::lifecycle::lifecycle::ObjectOpts;
pub(crate) type ReplicationConfig = ecstore_bucket::replication::ReplicationConfig;
pub(crate) type ReplicationHealQueueResult = ecstore_bucket::replication::ReplicationHealQueueResult;
pub(crate) type ReplicationQueueAdmission = ecstore_bucket::replication::ReplicationQueueAdmission;
pub(crate) type ScanGuard = ecstore_disk::ScanGuard;
pub(crate) type SetDisks = ecstore_set_disk::SetDisks;
pub(crate) type StorageError = ecstore_error::StorageError;
@@ -133,6 +137,39 @@ impl ScannerVersioningConfigExt for s3s::dto::VersioningConfiguration {
}
}
pub(crate) trait ScannerDiskExt {
async fn disk_info(&self, opts: &DiskInfoOptions) -> DiskResult<ecstore_disk::DiskInfo>;
async fn read_metadata(&self, volume: &str, path: &str) -> DiskResult<ecstore_disk::Bytes>;
fn path(&self) -> PathBuf;
fn get_disk_location(&self) -> DiskLocation;
fn start_scan(&self) -> ScanGuard;
}
impl<T> ScannerDiskExt for T
where
T: ecstore_disk::DiskAPI,
{
async fn disk_info(&self, opts: &DiskInfoOptions) -> DiskResult<ecstore_disk::DiskInfo> {
ecstore_disk::DiskAPI::disk_info(self, opts).await
}
async fn read_metadata(&self, volume: &str, path: &str) -> DiskResult<ecstore_disk::Bytes> {
ecstore_disk::DiskAPI::read_metadata(self, volume, path).await
}
fn path(&self) -> PathBuf {
ecstore_disk::DiskAPI::path(self)
}
fn get_disk_location(&self) -> DiskLocation {
ecstore_disk::DiskAPI::get_disk_location(self)
}
fn start_scan(&self) -> ScanGuard {
ecstore_disk::DiskAPI::start_scan(self)
}
}
pub(crate) async fn apply_transition_rule(event: &Event, src: &LcEventSrc, oi: &ScannerObjectInfo) -> bool {
ecstore_bucket::lifecycle::bucket_lifecycle_ops::apply_transition_rule(event, src, oi).await
}
@@ -41,8 +41,22 @@ pub(crate) type TierConfigMgr = ecstore_tier::tier::TierConfigMgr;
pub(crate) type TierMinIO = ecstore_tier::tier_config::TierMinIO;
pub(crate) type TierType = ecstore_tier::tier_config::TierType;
pub(crate) type TransitionOptions = ecstore_bucket::lifecycle::lifecycle::TransitionOptions;
pub(crate) use ecstore_tier::warm_backend::WarmBackend as ScannerWarmBackend;
pub(crate) type WarmBackendGetOpts = ecstore_tier::warm_backend::WarmBackendGetOpts;
pub(crate) trait ScannerTestDiskExt {
async fn read_metadata(&self, volume: &str, path: &str) -> ecstore_disk::error::Result<ecstore_disk::Bytes>;
}
impl<T> ScannerTestDiskExt for T
where
T: ecstore_disk::DiskAPI,
{
async fn read_metadata(&self, volume: &str, path: &str) -> ecstore_disk::error::Result<ecstore_disk::Bytes> {
ecstore_disk::DiskAPI::read_metadata(self, volume, path).await
}
}
#[allow(non_snake_case)]
pub(crate) fn EndpointServerPools(pools: Vec<PoolEndpoints>) -> EndpointServerPools {
ecstore_layout::EndpointServerPools::from(pools)
@@ -16,15 +16,13 @@ mod common;
use crate::common::storage_compat::{
BUCKET_LIFECYCLE_CONFIG, BucketVersioningSys, DiskOption, ECStore, Endpoint, EndpointServerPools, Endpoints,
GLOBAL_TierConfigMgr, PoolEndpoints, ReadCloser, ReaderImpl, STORAGE_FORMAT_FILE, TierConfig, TierMinIO, TierType,
TransitionOptions, WarmBackendGetOpts, build_transition_put_options, enqueue_transition_for_existing_objects,
get_bucket_metadata, init_background_expiry, init_bucket_metadata_sys, init_local_disks, new_disk,
path2_bucket_object_with_base_path, update_bucket_metadata,
GLOBAL_TierConfigMgr, PoolEndpoints, ReadCloser, ReaderImpl, STORAGE_FORMAT_FILE, ScannerTestDiskExt as _,
ScannerWarmBackend, TierConfig, TierMinIO, TierType, TransitionOptions, WarmBackendGetOpts, build_transition_put_options,
enqueue_transition_for_existing_objects, get_bucket_metadata, init_background_expiry, init_bucket_metadata_sys,
init_local_disks, new_disk, path2_bucket_object_with_base_path, update_bucket_metadata,
};
use futures::FutureExt;
use rustfs_config::ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT;
use rustfs_ecstore::api::disk::DiskAPI as _;
use rustfs_ecstore::api::tier::warm_backend::WarmBackend;
use rustfs_filemeta::FileMeta;
use rustfs_scanner::scanner_folder::ScannerItem;
use rustfs_scanner::scanner_io::ScannerIODisk;
@@ -687,7 +685,7 @@ impl MockWarmBackend {
}
#[async_trait::async_trait]
impl WarmBackend for MockWarmBackend {
impl ScannerWarmBackend for MockWarmBackend {
async fn put(&self, object: &str, r: ReaderImpl, _length: i64) -> Result<String, std::io::Error> {
let bytes = self.read_bytes(r).await?;
Ok(self.put_bytes(object, bytes, HashMap::new()).await)
+4 -3
View File
@@ -200,9 +200,10 @@ same pattern, keeping raw ECStore facade access centralized behind local
Outer bucket lifecycle, replication, versioning, object-lock, and
restore-request trait method access must stay behind local compatibility traits
or wrapper functions. Non-compat sources must not import those ECStore bucket
API traits directly after the wrapper boundary is established. Remaining
temporary direct ECStore method-resolution imports are limited to disk, RPC peer
client, and warm-backend traits until their hot-path wrappers are isolated.
API traits directly after the wrapper boundary is established. Disk, RPC peer
client, and warm-backend method-resolution access must follow the same pattern:
non-compat sources use owner-local compatibility traits or test aliases instead
of importing ECStore traits directly.
Scanner, notify, observability, and e2e `storage_compat.rs` boundaries must
also stay narrow. Scanner must not restore grouped bucket compatibility exports
for target, lifecycle, metadata, replication, or versioning modules. Notify
+36 -11
View File
@@ -5,17 +5,17 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block
## Current Context
- Issue: [`rustfs/backlog#660`](https://github.com/rustfs/backlog/issues/660)
- Branch: `overtrue/arch-compat-trait-method-wrappers`
- Baseline: completed `C-011/C-012/C-013/API-055/API-059/API-079/API-080/API-081/API-082/API-083/API-084/API-085/API-086/API-087/API-088/API-089/API-090/API-091/API-092/API-093/API-094/API-095`.
- Stacked on: `overtrue/arch-root-e2e-compat-raw-facade-prune` pending API-095 merge.
- Branch: `overtrue/arch-disk-rpc-method-wrappers`
- Baseline: completed `C-011/C-012/C-013/API-055/API-059/API-079/API-080/API-081/API-082/API-083/API-084/API-085/API-086/API-087/API-088/API-089/API-090/API-091/API-092/API-093/API-094/API-095/API-096`.
- Stacked on: `overtrue/arch-compat-trait-method-wrappers` pending API-096 merge.
- PR type for this branch: `pure-move`
- Runtime behavior changes: none.
- Rust code changes: move remaining outer bucket lifecycle, replication,
versioning, object-lock, and restore-request trait method access behind
local compatibility traits and wrapper functions.
- CI/script changes: stop allowing those bucket trait imports as direct ECStore
- Rust code changes: move remaining disk RPC, peer S3 RPC, heal/scanner disk,
and warm-backend test method access behind local compatibility traits or
aliases.
- CI/script changes: remove the final direct ECStore method-resolution import
exceptions outside compatibility boundaries.
- Docs changes: record the API-096 trait-method wrapper cleanup.
- Docs changes: record the API-097 disk/RPC/warm-backend wrapper cleanup.
## Phase 0 Tasks
@@ -376,6 +376,20 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block
- Verification: RustFS/scanner/heal compile coverage, direct bucket trait
import residual scan, migration guard, formatting, diff hygiene, Rust risk
scan, pre-commit quality gate, and three-expert review.
- [x] `API-097` Prune disk/RPC/warm-backend method imports.
- Current slice: move disk RPC, peer S3 RPC, heal/scanner disk, and
warm-backend test method access behind local compatibility traits or
aliases in the owning boundaries.
- Acceptance: non-compat RustFS, scanner, heal, and test sources no longer
import ECStore `DiskAPI`, `PeerS3Client`, or `WarmBackend` traits directly;
the migration guard no longer allowlists those direct imports.
- Must preserve: disk RPC request/response behavior, internode HTTP file and
walk streams, heal resume and auto-scan disk handling, scanner disk scan
behavior, and transition warm-backend test harness behavior.
- Verification: RustFS/scanner/heal/e2e compile coverage, direct
disk/RPC/warm-backend trait import residual scan, migration guard,
formatting, diff hygiene, Rust risk scan, pre-commit quality gate, and
three-expert review.
- [x] `G-012` Inventory placement and repair invariants.
- Acceptance:
[`placement-repair-invariants.md`](placement-repair-invariants.md) records
@@ -3409,14 +3423,25 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block
| Expert | Status | Notes |
|---|---|---|
| Quality/architecture | pass | API-096 localizes bucket trait method access behind owner-local compatibility traits/wrappers without re-exporting ECStore traits or widening the facade. |
| Migration preservation | pass | App/admin/storage/scanner call sites keep the same method names or direct wrapper calls; disk/RPC/warm-backend exceptions are unchanged. |
| Testing/verification | pass | Focused compile, residual scan, migration guard, formatting, diff hygiene, added-line risk scan, and full pre-commit passed. |
| Quality/architecture | pass | API-097 localizes disk/RPC/warm-backend method access behind owner-local compatibility traits and aliases without widening non-compat imports. |
| Migration preservation | pass | RPC, heal, scanner, and transition-test call sites keep existing behavior while losing direct ECStore trait imports. |
| Testing/verification | pass | Focused compile, residual scan, migration guard, layer guard, formatting, diff hygiene, risk scan, and full pre-commit passed. |
## Verification Notes
Passed before push:
- Issue #660 API-097 current slice:
- `cargo check -p rustfs -p rustfs-scanner -p rustfs-heal -p e2e_test --tests`: passed.
- `cargo fmt --all --check`: passed.
- `git diff --check`: passed.
- `bash -n scripts/check_architecture_migration_rules.sh`: passed.
- `./scripts/check_architecture_migration_rules.sh`: passed.
- `./scripts/check_layer_dependencies.sh`: passed.
- Direct non-compat disk/RPC/warm-backend trait import residual scan: passed.
- Rust risk scan on changed Rust files and guard script: passed.
- `make pre-commit`: passed.
- Issue #660 API-096 current slice:
- `cargo check -p rustfs -p rustfs-scanner -p rustfs-heal`: passed.
- `cargo fmt --all --check`: passed.
@@ -15,7 +15,7 @@
use super::{multipart_usecase::DefaultMultipartUsecase, object_usecase::DefaultObjectUsecase};
use crate::app::bucket_usecase::DefaultBucketUsecase;
use crate::app::storage_compat::{
ECStore, Endpoint, EndpointServerPools, Endpoints, GLOBAL_TierConfigMgr, PoolEndpoints, TierConfig, TierType,
AppWarmBackend, ECStore, Endpoint, EndpointServerPools, Endpoints, GLOBAL_TierConfigMgr, PoolEndpoints, TierConfig, TierType,
WarmBackendGetOpts,
metadata::{BUCKET_LIFECYCLE_CONFIG, OBJECT_LOCK_CONFIG},
metadata_sys,
@@ -31,7 +31,6 @@ use futures::FutureExt;
use futures::stream;
use http::{Extensions, HeaderMap, HeaderValue, Method, Uri, header::IF_NONE_MATCH};
use rustfs_config::{ENV_OBJECT_LOCK_OPTIMIZATION_ENABLE, ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT};
use rustfs_ecstore::api::tier::warm_backend::WarmBackend;
use rustfs_object_capacity::capacity_manager::{HybridStrategyConfig, create_isolated_manager};
use rustfs_storage_api::{
BucketOperations, BucketOptions, ListOperations as _, MakeBucketOptions, MultipartOperations as _, ObjectIO as _,
@@ -295,7 +294,7 @@ impl MockWarmBackend {
}
#[async_trait::async_trait]
impl WarmBackend for MockWarmBackend {
impl AppWarmBackend for MockWarmBackend {
async fn put(&self, object: &str, r: ReaderImpl, _length: i64) -> Result<String, std::io::Error> {
let bytes = self.read_bytes(r).await?;
Ok(self.put_bytes(object, bytes).await)
+2
View File
@@ -57,6 +57,8 @@ pub(crate) type TierConfig = crate::app::storage_compat::ecstore_tier::tier_conf
#[cfg(test)]
pub(crate) type TierType = crate::app::storage_compat::ecstore_tier::tier_config::TierType;
#[cfg(test)]
pub(crate) use crate::app::storage_compat::ecstore_tier::warm_backend::WarmBackend as AppWarmBackend;
#[cfg(test)]
pub(crate) type WarmBackendGetOpts = crate::app::storage_compat::ecstore_tier::warm_backend::WarmBackendGetOpts;
#[cfg(test)]
-1
View File
@@ -13,7 +13,6 @@
// limitations under the License.
use super::*;
use rustfs_ecstore::api::rpc::PeerS3Client as _;
impl NodeService {
pub(super) async fn handle_delete_bucket_metadata(
+1 -1
View File
@@ -15,6 +15,7 @@
use crate::server::RPC_PREFIX;
use crate::storage::request_context::spawn_traced;
use crate::storage::storage_compat::DEFAULT_READ_BUFFER_SIZE;
use crate::storage::storage_compat::StorageDiskRpcExt as _;
use crate::storage::storage_compat::WalkDirOptions;
use crate::storage::storage_compat::find_local_disk_by_ref;
use crate::storage::storage_compat::verify_rpc_signature;
@@ -24,7 +25,6 @@ use http::{HeaderMap, Method, Request, Response, StatusCode, Uri};
use http_body_util::{BodyExt, Limited};
use hyper::body::Incoming;
use rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE;
use rustfs_ecstore::api::disk::DiskAPI as _;
use rustfs_io_metrics::internode_metrics::{
INTERNODE_OPERATION_PUT_FILE_STREAM, INTERNODE_OPERATION_READ_FILE_STREAM, INTERNODE_OPERATION_WALK_DIR,
INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, global_internode_metrics,
+3 -4
View File
@@ -19,16 +19,15 @@ use crate::admin::service::{
use crate::storage::storage_compat::{
CollectMetricsOpts, DeleteOptions, DiskError, DiskInfoOptions, DiskStore, FileInfoVersions, LocalPeerS3Client, MetricType,
PEER_RESTSIGNAL, PEER_RESTSUB_SYS, ReadMultipleReq, ReadMultipleResp, ReadOptions, SERVICE_SIGNAL_REFRESH_CONFIG,
SERVICE_SIGNAL_RELOAD_DYNAMIC, UpdateMetadataOpts, all_local_disk_path, collect_local_metrics, find_local_disk_by_ref,
get_global_lock_client, get_local_server_property, load_bucket_metadata, reload_transition_tier_config,
resolve_object_store_handle, set_bucket_metadata,
SERVICE_SIGNAL_RELOAD_DYNAMIC, StorageDiskRpcExt as _, StoragePeerS3ClientExt as _, UpdateMetadataOpts, all_local_disk_path,
collect_local_metrics, find_local_disk_by_ref, get_global_lock_client, get_local_server_property, load_bucket_metadata,
reload_transition_tier_config, resolve_object_store_handle, set_bucket_metadata,
};
use bytes::Bytes;
use futures::Stream;
use futures_util::future::join_all;
use rmp_serde::Deserializer;
use rustfs_common::{get_global_local_node_name, heal_channel::HealOpts};
use rustfs_ecstore::api::disk::DiskAPI as _;
use rustfs_filemeta::{FileInfo, MetacacheReader};
use rustfs_iam::{get_global_iam_sys, store::UserType};
use rustfs_lock::{LockClient, LockRequest};
+294
View File
@@ -40,24 +40,33 @@ pub(crate) type BucketMetadata = ecstore_bucket::metadata::BucketMetadata;
#[cfg(test)]
pub(crate) type BucketMetadataSys = ecstore_bucket::metadata_sys::BucketMetadataSys;
pub(crate) type BucketVersioningSys = ecstore_bucket::versioning_sys::BucketVersioningSys;
pub(crate) type CheckPartsResp = ecstore_disk::CheckPartsResp;
pub(crate) type CollectMetricsOpts = ecstore_metrics::CollectMetricsOpts;
pub(crate) type DeleteOptions = ecstore_disk::DeleteOptions;
pub(crate) type DiskError = ecstore_disk::error::DiskError;
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;
pub(crate) type ECStore = ecstore_storage::ECStore;
pub(crate) type FileInfoVersions = ecstore_disk::FileInfoVersions;
pub(crate) type FileReader = ecstore_disk::FileReader;
pub(crate) type FileWriter = ecstore_disk::FileWriter;
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 PolicySys = ecstore_bucket::policy_sys::PolicySys;
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 StorageError = ecstore_error::StorageError;
pub(crate) type Error = StorageError;
pub(crate) type Result<T> = core::result::Result<T, Error>;
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;
@@ -65,6 +74,291 @@ pub(crate) async fn get_local_server_property() -> rustfs_madmin::ServerProperti
ecstore_admin::get_local_server_property().await
}
pub(crate) trait StorageDiskRpcExt {
async fn disk_info(&self, opts: &DiskInfoOptions) -> DiskResult<DiskInfo>;
async fn delete_volume(&self, volume: &str) -> DiskResult<()>;
async fn read_multiple(&self, req: ReadMultipleReq) -> DiskResult<Vec<ReadMultipleResp>>;
async fn delete_versions(&self, volume: &str, versions: Vec<FileInfoVersions>, opts: DeleteOptions)
-> Vec<Option<DiskError>>;
async fn delete_version(
&self,
volume: &str,
path: &str,
file_info: rustfs_filemeta::FileInfo,
force_del_marker: bool,
opts: DeleteOptions,
) -> DiskResult<()>;
async fn read_xl(&self, volume: &str, path: &str, read_data: bool) -> DiskResult<RawFileInfo>;
async fn read_version(
&self,
org_volume: &str,
volume: &str,
path: &str,
version_id: &str,
opts: &ReadOptions,
) -> DiskResult<rustfs_filemeta::FileInfo>;
async fn write_metadata(
&self,
org_volume: &str,
volume: &str,
path: &str,
file_info: rustfs_filemeta::FileInfo,
) -> DiskResult<()>;
async fn update_metadata(
&self,
volume: &str,
path: &str,
file_info: rustfs_filemeta::FileInfo,
opts: &UpdateMetadataOpts,
) -> DiskResult<()>;
async fn read_metadata(&self, volume: &str, path: &str) -> DiskResult<bytes::Bytes>;
async fn delete_paths(&self, volume: &str, paths: &[String]) -> DiskResult<()>;
async fn stat_volume(&self, volume: &str) -> DiskResult<VolumeInfo>;
async fn list_volumes(&self) -> DiskResult<Vec<VolumeInfo>>;
async fn make_volume(&self, volume: &str) -> DiskResult<()>;
async fn make_volumes(&self, volume: Vec<&str>) -> DiskResult<()>;
async fn rename_data(
&self,
src_volume: &str,
src_path: &str,
file_info: rustfs_filemeta::FileInfo,
dst_volume: &str,
dst_path: &str,
) -> DiskResult<RenameDataResp>;
async fn list_dir(&self, origvolume: &str, volume: &str, dir_path: &str, count: i32) -> DiskResult<Vec<String>>;
async fn read_file_stream(&self, volume: &str, path: &str, offset: usize, length: usize) -> DiskResult<FileReader>;
async fn rename_file(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str) -> DiskResult<()>;
async fn rename_part(
&self,
src_volume: &str,
src_path: &str,
dst_volume: &str,
dst_path: &str,
meta: bytes::Bytes,
) -> DiskResult<()>;
async fn delete(&self, volume: &str, path: &str, options: DeleteOptions) -> DiskResult<()>;
async fn verify_file(&self, volume: &str, path: &str, file_info: &rustfs_filemeta::FileInfo) -> DiskResult<CheckPartsResp>;
async fn check_parts(&self, volume: &str, path: &str, file_info: &rustfs_filemeta::FileInfo) -> DiskResult<CheckPartsResp>;
async fn read_parts(&self, bucket: &str, paths: &[String]) -> DiskResult<Vec<ObjectPartInfo>>;
async fn walk_dir<W: tokio::io::AsyncWrite + Unpin + Send>(&self, opts: WalkDirOptions, wr: &mut W) -> DiskResult<()>;
async fn write_all(&self, volume: &str, path: &str, data: bytes::Bytes) -> DiskResult<()>;
async fn read_all(&self, volume: &str, path: &str) -> DiskResult<bytes::Bytes>;
async fn append_file(&self, volume: &str, path: &str) -> DiskResult<FileWriter>;
async fn create_file(&self, origvolume: &str, volume: &str, path: &str, file_size: i64) -> DiskResult<FileWriter>;
}
impl<T> StorageDiskRpcExt for T
where
T: ecstore_disk::DiskAPI,
{
async fn disk_info(&self, opts: &DiskInfoOptions) -> DiskResult<DiskInfo> {
ecstore_disk::DiskAPI::disk_info(self, opts).await
}
async fn delete_volume(&self, volume: &str) -> DiskResult<()> {
ecstore_disk::DiskAPI::delete_volume(self, volume).await
}
async fn read_multiple(&self, req: ReadMultipleReq) -> DiskResult<Vec<ReadMultipleResp>> {
ecstore_disk::DiskAPI::read_multiple(self, req).await
}
async fn delete_versions(
&self,
volume: &str,
versions: Vec<FileInfoVersions>,
opts: DeleteOptions,
) -> Vec<Option<DiskError>> {
ecstore_disk::DiskAPI::delete_versions(self, volume, versions, opts).await
}
async fn delete_version(
&self,
volume: &str,
path: &str,
file_info: rustfs_filemeta::FileInfo,
force_del_marker: bool,
opts: DeleteOptions,
) -> DiskResult<()> {
ecstore_disk::DiskAPI::delete_version(self, volume, path, file_info, force_del_marker, opts).await
}
async fn read_xl(&self, volume: &str, path: &str, read_data: bool) -> DiskResult<RawFileInfo> {
ecstore_disk::DiskAPI::read_xl(self, volume, path, read_data).await
}
async fn read_version(
&self,
org_volume: &str,
volume: &str,
path: &str,
version_id: &str,
opts: &ReadOptions,
) -> DiskResult<rustfs_filemeta::FileInfo> {
ecstore_disk::DiskAPI::read_version(self, org_volume, volume, path, version_id, opts).await
}
async fn write_metadata(
&self,
org_volume: &str,
volume: &str,
path: &str,
file_info: rustfs_filemeta::FileInfo,
) -> DiskResult<()> {
ecstore_disk::DiskAPI::write_metadata(self, org_volume, volume, path, file_info).await
}
async fn update_metadata(
&self,
volume: &str,
path: &str,
file_info: rustfs_filemeta::FileInfo,
opts: &UpdateMetadataOpts,
) -> DiskResult<()> {
ecstore_disk::DiskAPI::update_metadata(self, volume, path, file_info, opts).await
}
async fn read_metadata(&self, volume: &str, path: &str) -> DiskResult<bytes::Bytes> {
ecstore_disk::DiskAPI::read_metadata(self, volume, path).await
}
async fn delete_paths(&self, volume: &str, paths: &[String]) -> DiskResult<()> {
ecstore_disk::DiskAPI::delete_paths(self, volume, paths).await
}
async fn stat_volume(&self, volume: &str) -> DiskResult<VolumeInfo> {
ecstore_disk::DiskAPI::stat_volume(self, volume).await
}
async fn list_volumes(&self) -> DiskResult<Vec<VolumeInfo>> {
ecstore_disk::DiskAPI::list_volumes(self).await
}
async fn make_volume(&self, volume: &str) -> DiskResult<()> {
ecstore_disk::DiskAPI::make_volume(self, volume).await
}
async fn make_volumes(&self, volume: Vec<&str>) -> DiskResult<()> {
ecstore_disk::DiskAPI::make_volumes(self, volume).await
}
async fn rename_data(
&self,
src_volume: &str,
src_path: &str,
file_info: rustfs_filemeta::FileInfo,
dst_volume: &str,
dst_path: &str,
) -> DiskResult<RenameDataResp> {
ecstore_disk::DiskAPI::rename_data(self, src_volume, src_path, file_info, dst_volume, dst_path).await
}
async fn list_dir(&self, origvolume: &str, volume: &str, dir_path: &str, count: i32) -> DiskResult<Vec<String>> {
ecstore_disk::DiskAPI::list_dir(self, origvolume, volume, dir_path, count).await
}
async fn read_file_stream(&self, volume: &str, path: &str, offset: usize, length: usize) -> DiskResult<FileReader> {
ecstore_disk::DiskAPI::read_file_stream(self, volume, path, offset, length).await
}
async fn rename_file(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str) -> DiskResult<()> {
ecstore_disk::DiskAPI::rename_file(self, src_volume, src_path, dst_volume, dst_path).await
}
async fn rename_part(
&self,
src_volume: &str,
src_path: &str,
dst_volume: &str,
dst_path: &str,
meta: bytes::Bytes,
) -> DiskResult<()> {
ecstore_disk::DiskAPI::rename_part(self, src_volume, src_path, dst_volume, dst_path, meta).await
}
async fn delete(&self, volume: &str, path: &str, options: DeleteOptions) -> DiskResult<()> {
ecstore_disk::DiskAPI::delete(self, volume, path, options).await
}
async fn verify_file(&self, volume: &str, path: &str, file_info: &rustfs_filemeta::FileInfo) -> DiskResult<CheckPartsResp> {
ecstore_disk::DiskAPI::verify_file(self, volume, path, file_info).await
}
async fn check_parts(&self, volume: &str, path: &str, file_info: &rustfs_filemeta::FileInfo) -> DiskResult<CheckPartsResp> {
ecstore_disk::DiskAPI::check_parts(self, volume, path, file_info).await
}
async fn read_parts(&self, bucket: &str, paths: &[String]) -> DiskResult<Vec<ObjectPartInfo>> {
ecstore_disk::DiskAPI::read_parts(self, bucket, paths).await
}
async fn walk_dir<W: tokio::io::AsyncWrite + Unpin + Send>(&self, opts: WalkDirOptions, wr: &mut W) -> DiskResult<()> {
ecstore_disk::DiskAPI::walk_dir(self, opts, wr).await
}
async fn write_all(&self, volume: &str, path: &str, data: bytes::Bytes) -> DiskResult<()> {
ecstore_disk::DiskAPI::write_all(self, volume, path, data).await
}
async fn read_all(&self, volume: &str, path: &str) -> DiskResult<bytes::Bytes> {
ecstore_disk::DiskAPI::read_all(self, volume, path).await
}
async fn append_file(&self, volume: &str, path: &str) -> DiskResult<FileWriter> {
ecstore_disk::DiskAPI::append_file(self, volume, path).await
}
async fn create_file(&self, origvolume: &str, volume: &str, path: &str, file_size: i64) -> DiskResult<FileWriter> {
ecstore_disk::DiskAPI::create_file(self, origvolume, volume, path, file_size).await
}
}
pub(crate) trait StoragePeerS3ClientExt {
async fn heal_bucket(
&self,
bucket: &str,
opts: &rustfs_common::heal_channel::HealOpts,
) -> DiskResult<rustfs_madmin::heal_commands::HealResultItem>;
async fn make_bucket(&self, bucket: &str, opts: &rustfs_storage_api::MakeBucketOptions) -> DiskResult<()>;
async fn list_bucket(&self, opts: &rustfs_storage_api::BucketOptions) -> DiskResult<Vec<rustfs_storage_api::BucketInfo>>;
async fn delete_bucket(&self, bucket: &str, opts: &rustfs_storage_api::DeleteBucketOptions) -> DiskResult<()>;
async fn get_bucket_info(
&self,
bucket: &str,
opts: &rustfs_storage_api::BucketOptions,
) -> DiskResult<rustfs_storage_api::BucketInfo>;
}
impl StoragePeerS3ClientExt for LocalPeerS3Client {
async fn heal_bucket(
&self,
bucket: &str,
opts: &rustfs_common::heal_channel::HealOpts,
) -> DiskResult<rustfs_madmin::heal_commands::HealResultItem> {
ecstore_rpc::PeerS3Client::heal_bucket(self, bucket, opts).await
}
async fn make_bucket(&self, bucket: &str, opts: &rustfs_storage_api::MakeBucketOptions) -> DiskResult<()> {
ecstore_rpc::PeerS3Client::make_bucket(self, bucket, opts).await
}
async fn list_bucket(&self, opts: &rustfs_storage_api::BucketOptions) -> DiskResult<Vec<rustfs_storage_api::BucketInfo>> {
ecstore_rpc::PeerS3Client::list_bucket(self, opts).await
}
async fn delete_bucket(&self, bucket: &str, opts: &rustfs_storage_api::DeleteBucketOptions) -> DiskResult<()> {
ecstore_rpc::PeerS3Client::delete_bucket(self, bucket, opts).await
}
async fn get_bucket_info(
&self,
bucket: &str,
opts: &rustfs_storage_api::BucketOptions,
) -> DiskResult<rustfs_storage_api::BucketInfo> {
ecstore_rpc::PeerS3Client::get_bucket_info(self, bucket, opts).await
}
}
pub(crate) async fn load_bucket_metadata(api: Arc<ECStore>, bucket: &str) -> Result<BucketMetadata> {
ecstore_bucket::metadata::load_bucket_metadata(api, bucket).await
}
@@ -691,12 +691,7 @@ fi
--glob '!**/storage_compat.rs' \
--glob '!target/**' || true
) |
perl -ne '
next if /\buse rustfs_ecstore::api::disk::DiskAPI(?:\s+as\s+_)?;/;
next if /\buse rustfs_ecstore::api::rpc::PeerS3Client(?:\s+as\s+_)?;/;
next if /\buse rustfs_ecstore::api::tier::warm_backend::WarmBackend(?:\s+as\s+_)?;/;
print;
' >"$DIRECT_ECSTORE_IMPORT_HITS_FILE"
cat >"$DIRECT_ECSTORE_IMPORT_HITS_FILE"
if [[ -s "$DIRECT_ECSTORE_IMPORT_HITS_FILE" ]]; then
report_failure "direct rustfs_ecstore imports outside compatibility boundaries are forbidden: $(paste -sd '; ' "$DIRECT_ECSTORE_IMPORT_HITS_FILE")"