mirror of
https://github.com/rustfs/rustfs.git
synced 2026-07-27 16:48:58 +00:00
refactor(ecstore): route replication target types (#4114)
This commit is contained in:
Zhengchao An
@@ -12,7 +12,7 @@ and lifecycle/heal scheduling paths.
|
||||
| `config.rs` | Replication config helpers, rule matching, and tag filtering. | Uses replication-local tagging boundary and S3 DTOs directly. |
|
||||
| `datatypes.rs` | Replication status and operation DTOs. | Publicly re-exported through the ECStore replication facade. |
|
||||
| `replication_pool.rs` | Replication queue, worker pool, MRF persistence, bucket stats, and delete/object scheduling. | Depends on bucket target sys, bucket metadata sys, config storage, object API, runtime sources, notification state, and storage-api object IO contracts. |
|
||||
| `replication_resyncer.rs` | Object replication, delete replication, resync, MRF encode/decode, target calls, and multipart target upload paths. | Depends on bucket target clients, metadata/versioning systems, ECStore object readers/writers, disk paths, runtime sources, notification events, bandwidth reader wrapping, and SetDisks lock timing. |
|
||||
| `replication_resyncer.rs` | Object replication, delete replication, resync, MRF encode/decode, target calls, and multipart target upload paths. | Depends on target calls through the replication target boundary, metadata/versioning systems, ECStore object readers/writers, disk paths, runtime sources, notification events, bandwidth reader wrapping, and SetDisks lock timing. |
|
||||
| `replication_state.rs` | Replication queue/stat state and worker accounting. | Reads runtime sources and owns shared replication pool/stat state. |
|
||||
| `rule.rs` | Rule evaluation helpers for object replication options. | Depends on ECStore replication object option types. |
|
||||
| `mod.rs` | Compatibility re-export facade for the current ECStore owner. | Must stay stable until downstream scanner, lifecycle, heal, and metrics paths compile through replacement contracts. |
|
||||
@@ -22,7 +22,8 @@ and lifecycle/heal scheduling paths.
|
||||
| Contract | Responsibility | Current dependency to remove |
|
||||
|---|---|---|
|
||||
| `ReplicationStorage` | Object read/write/delete, object walk, metadata update, and target object IO. | Direct ECStore object API and storage-api object IO coupling in worker paths. |
|
||||
| `ReplicationMetadataStore` | Replication config, bucket targets, MRF/resync state, target reset headers, and status persistence. | Direct bucket target sys, metadata sys, versioning sys, config storage, and file metadata imports. |
|
||||
| `ReplicationMetadataStore` | Replication config, MRF/resync state, target reset headers, and status persistence. | Direct metadata sys, versioning sys, config storage, and file metadata imports. |
|
||||
| `ReplicationTargetStore` | Bucket target listing, target client lookup, target offline checks, and target operation option types. | Bucket target sys access and target operation types are concentrated in `replication_target_boundary.rs`. |
|
||||
| `ReplicationRuntime` | Worker pool, queue sizing, stats, bucket monitor, local node identity, cancellation, and admission state. | Direct runtime source/global access and shared replication pool/stat state. |
|
||||
| `ReplicationBandwidthLimiter` | Target reader wrapping for replication bandwidth accounting and throttling. | Direct bucket bandwidth reader imports from resyncer paths. |
|
||||
| `ReplicationEventSink` | Notification and audit events for skipped, failed, pending, and completed replication operations. | Direct event notification service calls from worker code. |
|
||||
|
||||
@@ -19,12 +19,12 @@ use super::replication_lock_boundary as lock_boundary;
|
||||
use super::replication_metadata_boundary as metadata_boundary;
|
||||
use super::replication_msgp_boundary::{read_msgp_ext8_time, skip_msgp_value, write_msgp_time};
|
||||
use super::replication_tagging_boundary as tagging_boundary;
|
||||
use super::replication_target_boundary as target_boundary;
|
||||
use super::replication_versioning_boundary as versioning_boundary;
|
||||
use super::runtime_boundary as runtime_sources;
|
||||
use crate::bucket::bucket_target_sys::{
|
||||
use super::replication_target_boundary;
|
||||
use super::replication_target_boundary::{
|
||||
AdvancedPutOptions, PutObjectOptions, PutObjectPartOptions, RemoveObjectOptions, TargetClient,
|
||||
};
|
||||
use super::replication_versioning_boundary as versioning_boundary;
|
||||
use super::runtime_boundary as runtime_sources;
|
||||
use crate::bucket::replication::ResyncStatusType;
|
||||
use crate::bucket::replication::{ObjectOpts, ReplicationConfigurationExt as _};
|
||||
use crate::bucket::target::BucketTargets;
|
||||
@@ -875,7 +875,7 @@ impl ReplicationResyncer {
|
||||
}
|
||||
};
|
||||
|
||||
let targets = match target_boundary::list_bucket_targets(&opts.bucket).await {
|
||||
let targets = match replication_target_boundary::list_bucket_targets(&opts.bucket).await {
|
||||
Ok(targets) => targets,
|
||||
Err(err) => {
|
||||
debug!(
|
||||
@@ -920,7 +920,7 @@ impl ReplicationResyncer {
|
||||
return;
|
||||
}
|
||||
|
||||
let Some(target_client) = target_boundary::remote_target_client(&opts.bucket, &target_arns[0]).await else {
|
||||
let Some(target_client) = replication_target_boundary::remote_target_client(&opts.bucket, &target_arns[0]).await else {
|
||||
error!(
|
||||
event = EVENT_RESYNC_RUNTIME_SKIPPED,
|
||||
component = LOG_COMPONENT_ECSTORE,
|
||||
@@ -1675,7 +1675,7 @@ pub async fn check_replicate_delete(
|
||||
continue;
|
||||
}
|
||||
|
||||
let tgt = target_boundary::remote_target_client(bucket, &tgt_arn).await;
|
||||
let tgt = replication_target_boundary::remote_target_client(bucket, &tgt_arn).await;
|
||||
// The target online status should not be used here while deciding
|
||||
// whether to replicate deletes as the target could be temporarily down
|
||||
let tgt_dsc = if let Some(tgt) = tgt {
|
||||
@@ -1800,7 +1800,7 @@ pub async fn must_replicate(bucket: &str, object: &str, mopts: MustReplicateOpti
|
||||
let mut dsc = ReplicateDecision::default();
|
||||
|
||||
for arn in arns {
|
||||
let cli = target_boundary::remote_target_client(bucket, &arn).await;
|
||||
let cli = replication_target_boundary::remote_target_client(bucket, &arn).await;
|
||||
|
||||
let mut sopts = opts.clone();
|
||||
sopts.target_arn = arn.clone();
|
||||
@@ -2067,7 +2067,7 @@ pub async fn replicate_delete<S: ReplicationStorage>(dobj: DeletedObjectReplicat
|
||||
}
|
||||
|
||||
// Get the remote target client
|
||||
let Some(tgt_client) = target_boundary::remote_target_client(&bucket, &tgt_entry.arn).await else {
|
||||
let Some(tgt_client) = replication_target_boundary::remote_target_client(&bucket, &tgt_entry.arn).await else {
|
||||
debug!(
|
||||
event = EVENT_REPLICATION_DELETE_SKIPPED,
|
||||
component = LOG_COMPONENT_ECSTORE,
|
||||
@@ -2293,7 +2293,7 @@ async fn replicate_delete_marker_purge_to_targets(bucket: &str, dobj: &DeletedOb
|
||||
if !dobj.target_arn.is_empty() && dobj.target_arn != tgt_entry.arn {
|
||||
continue;
|
||||
}
|
||||
let Some(tgt_client) = target_boundary::remote_target_client(bucket, &tgt_entry.arn).await else {
|
||||
let Some(tgt_client) = replication_target_boundary::remote_target_client(bucket, &tgt_entry.arn).await else {
|
||||
continue;
|
||||
};
|
||||
|
||||
@@ -2444,7 +2444,7 @@ async fn replicate_force_delete_to_targets<S: ReplicationStorage>(dobj: &Deleted
|
||||
let mut join_set = JoinSet::new();
|
||||
|
||||
for arn in tgt_arns {
|
||||
let Some(tgt_client) = target_boundary::remote_target_client(bucket, &arn).await else {
|
||||
let Some(tgt_client) = replication_target_boundary::remote_target_client(bucket, &arn).await else {
|
||||
debug!(
|
||||
event = EVENT_REPLICATION_FORCE_DELETE_SKIPPED,
|
||||
component = LOG_COMPONENT_ECSTORE,
|
||||
@@ -2473,7 +2473,7 @@ async fn replicate_force_delete_to_targets<S: ReplicationStorage>(dobj: &Deleted
|
||||
let object_name = object_name.clone();
|
||||
|
||||
join_set.spawn(async move {
|
||||
if target_boundary::target_is_offline(&tgt_client).await {
|
||||
if replication_target_boundary::target_is_offline(&tgt_client).await {
|
||||
error!(
|
||||
event = EVENT_REPLICATION_FORCE_DELETE_SKIPPED,
|
||||
component = LOG_COMPONENT_ECSTORE,
|
||||
@@ -2601,7 +2601,7 @@ async fn replicate_delete_to_target(dobj: &DeletedObjectReplicationInfo, tgt_cli
|
||||
return rinfo;
|
||||
}
|
||||
|
||||
if target_boundary::target_is_offline(&tgt_client).await {
|
||||
if replication_target_boundary::target_is_offline(&tgt_client).await {
|
||||
if !is_version_purge {
|
||||
rinfo.replication_status = ReplicationStatusType::Failed;
|
||||
} else {
|
||||
@@ -2830,7 +2830,7 @@ pub async fn replicate_object<S: ReplicationStorage>(roi: ReplicateObjectInfo, s
|
||||
let mut join_set = JoinSet::new();
|
||||
|
||||
for arn in tgt_arns {
|
||||
let Some(tgt_client) = target_boundary::remote_target_client(&bucket, &arn).await else {
|
||||
let Some(tgt_client) = replication_target_boundary::remote_target_client(&bucket, &arn).await else {
|
||||
debug!(
|
||||
event = EVENT_RESYNC_RUNTIME_SKIPPED,
|
||||
component = LOG_COMPONENT_ECSTORE,
|
||||
@@ -2988,7 +2988,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
|
||||
return rinfo;
|
||||
}
|
||||
|
||||
if target_boundary::target_is_offline(&tgt_client).await {
|
||||
if replication_target_boundary::target_is_offline(&tgt_client).await {
|
||||
debug!(
|
||||
event = EVENT_RESYNC_RUNTIME_SKIPPED,
|
||||
component = LOG_COMPONENT_ECSTORE,
|
||||
@@ -3275,7 +3275,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
if target_boundary::target_is_offline(&tgt_client).await {
|
||||
if replication_target_boundary::target_is_offline(&tgt_client).await {
|
||||
debug!(
|
||||
event = EVENT_RESYNC_RUNTIME_SKIPPED,
|
||||
component = LOG_COMPONENT_ECSTORE,
|
||||
|
||||
@@ -14,9 +14,13 @@
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use crate::bucket::bucket_target_sys::{BucketTargetError, BucketTargetSys, TargetClient};
|
||||
use crate::bucket::bucket_target_sys::{BucketTargetError, BucketTargetSys};
|
||||
use crate::bucket::target::BucketTargets;
|
||||
|
||||
pub(crate) use crate::bucket::bucket_target_sys::{
|
||||
AdvancedPutOptions, PutObjectOptions, PutObjectPartOptions, RemoveObjectOptions, TargetClient,
|
||||
};
|
||||
|
||||
pub(crate) async fn list_bucket_targets(bucket: &str) -> Result<BucketTargets, BucketTargetError> {
|
||||
BucketTargetSys::get().list_bucket_targets(bucket).await
|
||||
}
|
||||
|
||||
@@ -106,8 +106,9 @@ Current coupling:
|
||||
scanner repair classification, runtime replication pool/stat handles, bucket
|
||||
monitor and bandwidth reader access through local boundaries, local node
|
||||
names, and notification events;
|
||||
- resync and delete replication paths call metadata and bucket target systems
|
||||
directly;
|
||||
- resync and delete replication paths call metadata directly, while bucket
|
||||
target system access and target operation types are concentrated behind the
|
||||
replication target boundary;
|
||||
- lifecycle and heal paths schedule replication work through the current ECStore
|
||||
module;
|
||||
- global replication pool/stat initialization still lives with ECStore runtime
|
||||
@@ -118,8 +119,10 @@ Required contracts before crate movement:
|
||||
- `ReplicationStorage`: keep the existing trait as the starting point, then
|
||||
split object read/write/delete, walk, and metadata update responsibilities
|
||||
only when call sites prove a narrower shape.
|
||||
- `ReplicationMetadataStore`: replication config, bucket targets, target reset
|
||||
headers, MRF/resync state, and status persistence.
|
||||
- `ReplicationMetadataStore`: replication config, target reset headers,
|
||||
MRF/resync state, and status persistence.
|
||||
- `ReplicationTargetStore`: bucket target listing, target client lookup,
|
||||
target offline checks, and target operation option types.
|
||||
- `ReplicationRuntime`: pool, stats, worker admission, bucket monitor, local
|
||||
node identity, cancellation, and queue sizing.
|
||||
- `ReplicationBandwidthLimiter`: target reader wrapping for replication
|
||||
|
||||
@@ -2590,14 +2590,21 @@ fi
|
||||
|
||||
(
|
||||
cd "$ROOT_DIR"
|
||||
rg -n --with-filename 'BucketTargetSys::get\(\)' \
|
||||
crates/ecstore/src/bucket/replication \
|
||||
--glob '*.rs' |
|
||||
find crates/ecstore/src/bucket/replication -type f -name '*.rs' -print0 |
|
||||
xargs -0 perl -0ne '
|
||||
while (/(?:(?:crate::bucket|super(?:::super)+)(?:::\{[^;]*\bbucket_target_sys\b|::bucket_target_sys\b)|BucketTargetSys::get\(\))/sg) {
|
||||
my $prefix = substr($_, 0, $-[0]);
|
||||
my $line = ($prefix =~ tr/\n//) + 1;
|
||||
my $match = $&;
|
||||
$match =~ s/\s+/ /g;
|
||||
print "$ARGV:$line:$match\n";
|
||||
}
|
||||
' |
|
||||
rg -v '^crates/ecstore/src/bucket/replication/replication_target_boundary\.rs:' || true
|
||||
) >"$REPLICATION_TARGET_BOUNDARY_BYPASS_HITS_FILE"
|
||||
|
||||
if [[ -s "$REPLICATION_TARGET_BOUNDARY_BYPASS_HITS_FILE" ]]; then
|
||||
report_failure "replication bucket-target access must stay behind replication target boundary: $(paste -sd '; ' "$REPLICATION_TARGET_BOUNDARY_BYPASS_HITS_FILE")"
|
||||
report_failure "replication bucket-target access and types must stay behind replication target boundary: $(paste -sd '; ' "$REPLICATION_TARGET_BOUNDARY_BYPASS_HITS_FILE")"
|
||||
fi
|
||||
|
||||
(
|
||||
|
||||
Reference in New Issue
Block a user