From f0adabbe9038d62fb241572cded23a2a8feb8a88 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Thu, 2 Jul 2026 16:32:02 +0800 Subject: [PATCH] refactor(replication): move object config bridge (#4188) * refactor(replication): move object config bridge * refactor(replication): centralize log constants --- crates/ecstore/src/bucket/replication/mod.rs | 4 +- .../replication_lifecycle_bridge.rs | 2 +- .../bucket/replication/replication_logging.rs | 19 + .../replication/replication_object_bridge.rs | 2 +- .../replication/replication_object_config.rs | 362 ++++++++++++++++++ .../bucket/replication/replication_pool.rs | 9 +- .../replication/replication_resyncer.rs | 312 +-------------- .../replication/replication_scanner_bridge.rs | 2 +- 8 files changed, 405 insertions(+), 307 deletions(-) create mode 100644 crates/ecstore/src/bucket/replication/replication_logging.rs create mode 100644 crates/ecstore/src/bucket/replication/replication_object_config.rs diff --git a/crates/ecstore/src/bucket/replication/mod.rs b/crates/ecstore/src/bucket/replication/mod.rs index fc2f978d2..6bcf6fa8c 100644 --- a/crates/ecstore/src/bucket/replication/mod.rs +++ b/crates/ecstore/src/bucket/replication/mod.rs @@ -21,10 +21,12 @@ mod replication_event_sink; mod replication_filemeta_boundary; mod replication_lifecycle_bridge; mod replication_lock_boundary; +mod replication_logging; mod replication_metadata_boundary; mod replication_migration_bridge; mod replication_msgp_boundary; mod replication_object_bridge; +mod replication_object_config; pub(crate) mod replication_pool; mod replication_resyncer; mod replication_scanner_bridge; @@ -41,11 +43,11 @@ pub use datatypes::ResyncStatusType; pub(crate) use replication_lifecycle_bridge::{ReplicationLifecycleBridge, ReplicationLifecycleConfig}; pub(crate) use replication_migration_bridge::ReplicationMigrationBridge; pub use replication_object_bridge::ReplicationObjectBridge; +pub use replication_object_config::ReplicationConfig; pub use replication_pool::{ DynReplicationPool, ReplicationPoolTrait, get_global_replication_pool, get_global_replication_stats, init_background_replication, }; -pub use replication_resyncer::ReplicationConfig; pub use replication_scanner_bridge::ReplicationScannerBridge; pub use replication_state::ReplicationStats; pub use replication_storage_boundary::{ReplicationObjectIO, ReplicationStorage}; diff --git a/crates/ecstore/src/bucket/replication/replication_lifecycle_bridge.rs b/crates/ecstore/src/bucket/replication/replication_lifecycle_bridge.rs index 98c6d1ff7..3d13cb751 100644 --- a/crates/ecstore/src/bucket/replication/replication_lifecycle_bridge.rs +++ b/crates/ecstore/src/bucket/replication/replication_lifecycle_bridge.rs @@ -18,7 +18,7 @@ use super::config::ReplicationConfigurationExt as _; use super::replication_filemeta_boundary::{ REPLICATE_INCOMING_DELETE, ReplicateDecision, ReplicationState, version_purge_statuses_map, }; -use super::replication_resyncer::{ReplicationConfig, check_replicate_delete}; +use super::replication_object_config::{ReplicationConfig, check_replicate_delete}; use super::replication_storage_boundary::{DeletedObject, ObjectInfo, ObjectOptions, ObjectToDelete}; use rustfs_replication::DeletedObjectReplicationInfo; diff --git a/crates/ecstore/src/bucket/replication/replication_logging.rs b/crates/ecstore/src/bucket/replication/replication_logging.rs new file mode 100644 index 000000000..c8d51ea23 --- /dev/null +++ b/crates/ecstore/src/bucket/replication/replication_logging.rs @@ -0,0 +1,19 @@ +// 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) const LOG_COMPONENT_ECSTORE: &str = "ecstore"; +pub(crate) const LOG_SUBSYSTEM_REPLICATION: &str = "replication"; +pub(crate) const LOG_SUBSYSTEM_REPLICATION_RESYNC: &str = "replication_resync"; +pub(crate) const EVENT_REPLICATION_CONFIG_LOOKUP_SKIPPED: &str = "replication_config_lookup_skipped"; +pub(crate) const EVENT_RESYNC_CONFIG_LOOKUP_SKIPPED: &str = "replication_resync_config_lookup_skipped"; diff --git a/crates/ecstore/src/bucket/replication/replication_object_bridge.rs b/crates/ecstore/src/bucket/replication/replication_object_bridge.rs index 8d2e6c81c..a512ac7a2 100644 --- a/crates/ecstore/src/bucket/replication/replication_object_bridge.rs +++ b/crates/ecstore/src/bucket/replication/replication_object_bridge.rs @@ -15,8 +15,8 @@ use std::{collections::HashMap, sync::Arc}; use super::replication_filemeta_boundary::{ReplicateDecision, ReplicationStatusType, ReplicationType}; +use super::replication_object_config::{check_replicate_delete, get_must_replicate_options, must_replicate}; use super::replication_pool::{schedule_replication, schedule_replication_delete}; -use super::replication_resyncer::{check_replicate_delete, get_must_replicate_options, must_replicate}; use super::replication_storage_boundary::{ObjectInfo, ObjectOptions, ObjectToDelete, ReplicationStorage}; use rustfs_replication::{DeletedObjectReplicationInfo, MustReplicateOptions}; diff --git a/crates/ecstore/src/bucket/replication/replication_object_config.rs b/crates/ecstore/src/bucket/replication/replication_object_config.rs new file mode 100644 index 000000000..eab0bd100 --- /dev/null +++ b/crates/ecstore/src/bucket/replication/replication_object_config.rs @@ -0,0 +1,362 @@ +// 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::collections::HashMap; + +use rustfs_replication::{ + MustReplicateOptions, ReplicationDeleteSource, ReplicationResyncTargetObject, delete_replication_missing_source_decision, + delete_replication_object_opts, resync_target_for_object, +}; +use rustfs_utils::http::AMZ_BUCKET_REPLICATION_STATUS; +use s3s::dto::ReplicationConfiguration; +use serde::{Deserialize, Serialize}; +use tracing::error; + +use super::config::{ObjectOpts, ReplicationConfigurationExt as _}; +use super::replication_error_boundary::Result; +use super::replication_filemeta_boundary::{ + ReplicateDecision, ReplicateTargetDecision, ReplicationStatusType, ReplicationType, ResyncDecision, +}; +use super::replication_logging::{EVENT_RESYNC_CONFIG_LOOKUP_SKIPPED, LOG_COMPONENT_ECSTORE, LOG_SUBSYSTEM_REPLICATION_RESYNC}; +use super::replication_metadata_boundary::ReplicationMetadataStore; +use super::replication_storage_boundary::{ObjectInfo, ObjectOptions, ObjectToDelete}; +use super::replication_target_boundary::{BucketTargets, ReplicationTargetStore}; +use super::replication_versioning_boundary::ReplicationVersioningStore; +use super::runtime_boundary as runtime_sources; + +pub(crate) async fn get_replication_config(bucket: &str) -> Result> { + ReplicationMetadataStore::optional_replication_config(bucket).await +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct ReplicationConfig { + pub config: Option, + pub remotes: Option, +} + +impl ReplicationConfig { + pub fn new(config: Option, remotes: Option) -> Self { + Self { config, remotes } + } + + pub fn is_empty(&self) -> bool { + self.config.is_none() + } + + pub fn replicate(&self, obj: &ObjectOpts) -> bool { + self.config.as_ref().is_some_and(|config| config.replicate(obj)) + } + + pub async fn resync( + &self, + oi: ObjectInfo, + dsc: ReplicateDecision, + status: &HashMap, + ) -> ResyncDecision { + if self.is_empty() { + return ResyncDecision::default(); + } + + let mut dsc = dsc; + + if oi.delete_marker { + let opts = ObjectOpts { + name: oi.name.clone(), + version_id: oi.version_id, + delete_marker: true, + op_type: ReplicationType::Delete, + existing_object: true, + ..Default::default() + }; + let arns = self + .config + .as_ref() + .map(|config| config.filter_target_arns(&opts)) + .unwrap_or_default(); + + if arns.is_empty() { + return ResyncDecision::default(); + } + + for arn in arns { + let mut opts = opts.clone(); + opts.target_arn = arn; + + dsc.set(ReplicateTargetDecision::new(opts.target_arn.clone(), self.replicate(&opts), false)); + } + + return self.resync_internal(oi, dsc, status); + } + + let mut user_defined = (*oi.user_defined).clone(); + user_defined.remove(AMZ_BUCKET_REPLICATION_STATUS); + + let dsc = must_replicate( + oi.bucket.as_str(), + &oi.name, + MustReplicateOptions::new(&user_defined, (*oi.user_tags).clone(), ReplicationType::ExistingObject, false), + ) + .await; + + self.resync_internal(oi, dsc, status) + } + + fn resync_internal( + &self, + oi: ObjectInfo, + dsc: ReplicateDecision, + status: &HashMap, + ) -> ResyncDecision { + let Some(remotes) = self.remotes.as_ref() else { + return ResyncDecision::default(); + }; + + if remotes.is_empty() { + return ResyncDecision::default(); + } + + let mut resync_decision = ResyncDecision::default(); + + for target in remotes.targets.iter() { + if let Some(decision) = dsc.targets_map.get(&target.arn) + && decision.replicate + { + resync_decision.targets.insert( + decision.arn.clone(), + resync_target_for_object( + &ReplicationResyncTargetObject { + mod_time: oi.mod_time, + user_defined: oi.user_defined.as_ref(), + }, + &target.arn, + &target.reset_id, + target.reset_before_date, + status.get(&decision.arn).unwrap_or(&ReplicationStatusType::Empty).clone(), + ), + ); + } + } + + resync_decision + } +} + +pub(crate) fn get_must_replicate_options( + user_defined: &HashMap, + user_tags: String, + _status: ReplicationStatusType, + op_type: ReplicationType, + opts: ObjectOptions, +) -> MustReplicateOptions { + MustReplicateOptions::new(user_defined, user_tags, op_type, opts.replication_request) +} + +pub(crate) async fn check_replicate_delete( + bucket: &str, + dobj: &ObjectToDelete, + oi: &ObjectInfo, + del_opts: &ObjectOptions, + gerr: Option, +) -> ReplicateDecision { + let rcfg = match get_replication_config(bucket).await { + Ok(Some(config)) => config, + Ok(None) => return ReplicateDecision::default(), + Err(err) => { + error!( + event = EVENT_RESYNC_CONFIG_LOOKUP_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + reason = "replication_config_lookup_failed", + error = %err, + "Failed to look up replication config for delete replication" + ); + return ReplicateDecision::default(); + } + }; + + if del_opts.replication_request { + return ReplicateDecision::default(); + } + + if !del_opts.versioned { + return ReplicateDecision::default(); + } + + let opts = delete_replication_object_opts( + dobj, + &ReplicationDeleteSource { + user_defined: oi.user_defined.as_ref(), + user_tags: oi.user_tags.as_str(), + delete_marker: oi.delete_marker, + replication_status: oi.replication_status.clone(), + }, + ); + + let tgt_arns = rcfg.filter_target_arns(&opts); + let mut dsc = ReplicateDecision::new(); + + if tgt_arns.is_empty() { + return dsc; + } + + for tgt_arn in tgt_arns { + let mut opts = opts.clone(); + opts.target_arn = tgt_arn.clone(); + let replicate = rcfg.replicate(&opts); + let sync = false; + + if gerr.is_some() { + if let Some(replicate) = delete_replication_missing_source_decision( + oi.delete_marker, + oi.target_replication_status(&tgt_arn), + replicate, + &oi.version_purge_status, + ) { + dsc.set(ReplicateTargetDecision::new(tgt_arn, replicate, sync)); + } + continue; + } + + let tgt = ReplicationTargetStore::remote_target_client(bucket, &tgt_arn).await; + let tgt_dsc = if let Some(tgt) = tgt { + ReplicateTargetDecision::new(tgt_arn, replicate, tgt.replicate_sync) + } else { + ReplicateTargetDecision::new(tgt_arn, false, false) + }; + dsc.set(tgt_dsc); + } + + dsc +} + +pub(crate) async fn must_replicate(bucket: &str, object: &str, mopts: MustReplicateOptions) -> ReplicateDecision { + if runtime_sources::object_store_handle().is_none() { + return ReplicateDecision::default(); + } + + if !ReplicationVersioningStore::prefix_enabled(bucket, object).await { + return ReplicateDecision::default(); + } + + let replication_status = mopts.replication_status(); + + if replication_status == ReplicationStatusType::Replica && !mopts.is_metadata_replication() { + return ReplicateDecision::default(); + } + + if mopts.is_replication_request() { + return ReplicateDecision::default(); + } + + let cfg = match get_replication_config(bucket).await { + Ok(Some(cfg)) => cfg, + Ok(None) | Err(_) => return ReplicateDecision::default(), + }; + + let opts = ObjectOpts { + name: object.to_string(), + replica: replication_status == ReplicationStatusType::Replica, + existing_object: mopts.is_existing_object_replication(), + user_tags: mopts.user_tags().to_string(), + ..Default::default() + }; + + let arns = cfg.filter_target_arns(&opts); + + if arns.is_empty() { + return ReplicateDecision::default(); + } + + let mut dsc = ReplicateDecision::default(); + + for arn in arns { + let cli = ReplicationTargetStore::remote_target_client(bucket, &arn).await; + + let mut sopts = opts.clone(); + sopts.target_arn = arn.clone(); + + let replicate = cfg.replicate(&sopts); + let synchronous = if let Some(cli) = cli { cli.replicate_sync } else { false }; + + dsc.set(ReplicateTargetDecision::new(arn, replicate, synchronous)); + } + + dsc +} + +#[cfg(test)] +mod tests { + use s3s::dto::{Destination, ReplicationRule, ReplicationRuleStatus}; + + use super::*; + + fn replication_rule() -> ReplicationRule { + ReplicationRule { + delete_marker_replication: None, + delete_replication: None, + destination: Destination { + bucket: "arn:aws:s3:::target-bucket".to_string(), + ..Default::default() + }, + existing_object_replication: None, + filter: None, + id: Some("rule".to_string()), + prefix: Some(String::new()), + priority: Some(1), + source_selection_criteria: None, + status: ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED), + } + } + + #[test] + fn replication_config_empty_and_replicate_follow_config() { + let empty = ReplicationConfig::default(); + assert!(empty.is_empty()); + assert!(!empty.replicate(&ObjectOpts::default())); + + let config = ReplicationConfig::new( + Some(ReplicationConfiguration { + role: String::new(), + rules: vec![replication_rule()], + }), + None, + ); + + assert!(!config.is_empty()); + assert!(config.replicate(&ObjectOpts { + name: "object".to_string(), + ..Default::default() + })); + } + + #[test] + fn must_replicate_options_preserve_request_flag() { + let user_defined = HashMap::new(); + let options = get_must_replicate_options( + &user_defined, + "env=prod".to_string(), + ReplicationStatusType::Empty, + ReplicationType::Metadata, + ObjectOptions { + replication_request: true, + ..Default::default() + }, + ); + + assert!(options.is_metadata_replication()); + assert!(options.is_replication_request()); + assert_eq!(options.user_tags(), "env=prod"); + } +} diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs index 4fcd15f77..024d209e7 100644 --- a/crates/ecstore/src/bucket/replication/replication_pool.rs +++ b/crates/ecstore/src/bucket/replication/replication_pool.rs @@ -20,10 +20,12 @@ use super::replication_filemeta_boundary::{ ReplicationStatusType, ReplicationType, ReplicationWorkerOperation, ResyncDecision, replication_statuses_map, version_purge_statuses_map, }; +use super::replication_logging::{EVENT_REPLICATION_CONFIG_LOOKUP_SKIPPED, LOG_COMPONENT_ECSTORE, LOG_SUBSYSTEM_REPLICATION}; use super::replication_metadata_boundary::ReplicationMetadataStore; +use super::replication_object_config::ReplicationConfig; use super::replication_resyncer::{ - ReplicationConfig, ReplicationResyncer, decode_mrf_file, decode_resync_file, encode_mrf_file, get_heal_replicate_object_info, - replicate_delete, replicate_object, save_resync_status, + ReplicationResyncer, decode_mrf_file, decode_resync_file, encode_mrf_file, get_heal_replicate_object_info, replicate_delete, + replicate_object, save_resync_status, }; use super::replication_state::ReplicationStats; use super::replication_storage_boundary::{DeletedObject, ObjectInfo, ObjectOptions, ReplicationObjectIO, ReplicationStorage}; @@ -54,14 +56,11 @@ use tokio::time::Duration; use tokio_util::sync::CancellationToken; use tracing::{debug, info, instrument, warn}; -const LOG_COMPONENT_ECSTORE: &str = "ecstore"; -const LOG_SUBSYSTEM_REPLICATION: &str = "replication"; const EVENT_REPLICATION_WORKER_RESIZE_SKIPPED: &str = "replication_worker_resize_skipped"; const EVENT_REPLICATION_WORKER_RESIZED: &str = "replication_worker_resized"; const EVENT_REPLICATION_BACKPRESSURE: &str = "replication_backpressure"; const EVENT_REPLICATION_RESYNC_LOAD_SKIPPED: &str = "replication_resync_load_skipped"; const EVENT_REPLICATION_RESYNC_RECOVERED: &str = "replication_resync_recovered"; -const EVENT_REPLICATION_CONFIG_LOOKUP_SKIPPED: &str = "replication_config_lookup_skipped"; const EVENT_REPLICATION_MRF_QUEUE_OVERFLOW: &str = "replication_mrf_queue_overflow"; /// Main replication pool structure diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index bcfb5e8f8..41b2cbc8b 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -19,23 +19,24 @@ use super::replication_config_store::ReplicationConfigStore; use super::replication_error_boundary::{Error, Result, is_err_object_not_found, is_err_version_not_found}; use super::replication_event_sink::{EventArgs, send_event, send_local_event}; use super::replication_filemeta_boundary::{ - MrfReplicateEntry, REPLICATE_EXISTING, REPLICATE_EXISTING_DELETE, ReplicateDecision, ReplicateObjectInfo, - ReplicateTargetDecision, ReplicatedInfos, ReplicatedTargetInfo, ReplicationAction, ReplicationStatusType, ReplicationType, - ResyncDecision, VersionPurgeStatusType, get_replication_state, parse_replicate_decision, replication_statuses_map, - target_reset_header, version_purge_statuses_map, + MrfReplicateEntry, REPLICATE_EXISTING, REPLICATE_EXISTING_DELETE, ReplicateDecision, ReplicateObjectInfo, ReplicatedInfos, + ReplicatedTargetInfo, ReplicationAction, ReplicationStatusType, ReplicationType, VersionPurgeStatusType, + get_replication_state, parse_replicate_decision, replication_statuses_map, target_reset_header, version_purge_statuses_map, }; use super::replication_lock_boundary::ReplicationLockTiming; +use super::replication_logging::{EVENT_RESYNC_CONFIG_LOOKUP_SKIPPED, LOG_COMPONENT_ECSTORE, LOG_SUBSYSTEM_REPLICATION_RESYNC}; use super::replication_metadata_boundary::ReplicationMetadataStore; #[cfg(test)] use super::replication_msgp_boundary::ReplicationMsgpCodec; +use super::replication_object_config::{ReplicationConfig, check_replicate_delete, get_replication_config, must_replicate}; use super::replication_storage_boundary::{ AdvancedGetOptions, DeletedObject, EcstoreObjectOperations, HTTPRangeSpec, ObjectInfo, ObjectOptions, ObjectToDelete, ReplicationObjectIO, ReplicationStorage, StatObjectOptions, WalkOptions, }; use super::replication_target_boundary::{ - BucketTargets, PutObjectOptions, PutObjectPartOptions, ReplicationTargetStore, TargetClient, - replication_complete_multipart_options, replication_delete_marker_purge_remove_options, replication_delete_remove_options, - replication_force_delete_remove_options, replication_put_object_header_size, replication_put_object_options, + PutObjectOptions, PutObjectPartOptions, ReplicationTargetStore, TargetClient, replication_complete_multipart_options, + replication_delete_marker_purge_remove_options, replication_delete_remove_options, replication_force_delete_remove_options, + replication_put_object_header_size, replication_put_object_options, }; use super::replication_versioning_boundary::ReplicationVersioningStore; use super::runtime_boundary as runtime_sources; @@ -54,22 +55,19 @@ use rmp_serde; #[cfg(test)] use rustfs_replication::content_matches_by_etag; use rustfs_replication::{ - BucketReplicationResyncStatus, DeletedObjectReplicationInfo, MustReplicateOptions, ReplicationDeleteSource, - ReplicationResyncTargetObject, ReplicationSourceObject, ReplicationTargetObject, ResyncOpts, TargetReplicationResyncStatus, - delete_replication_missing_source_decision, delete_replication_object_opts, heal_uses_delete_replication_path, + BucketReplicationResyncStatus, DeletedObjectReplicationInfo, MustReplicateOptions, ReplicationSourceObject, + ReplicationTargetObject, ResyncOpts, TargetReplicationResyncStatus, heal_uses_delete_replication_path, is_retryable_delete_replication_head_error, is_version_delete_replication, is_version_id_mismatch, - replication_action_for_target, replication_etags_match, resync_state_accepts_update, resync_target_for_object, - should_count_head_proxy_failure, should_retry_delete_marker_purge, target_is_newer_than_source_null_version, + replication_action_for_target, replication_etags_match, resync_state_accepts_update, should_count_head_proxy_failure, + should_retry_delete_marker_purge, target_is_newer_than_source_null_version, }; use rustfs_s3_types::EventName; use rustfs_utils::http::{ - AMZ_BUCKET_REPLICATION_STATUS, AMZ_TAGGING_DIRECTIVE, SUFFIX_REPLICATION_RESET, SUFFIX_REPLICATION_STATUS, - has_internal_suffix, insert_str, + AMZ_TAGGING_DIRECTIVE, SUFFIX_REPLICATION_RESET, SUFFIX_REPLICATION_STATUS, has_internal_suffix, insert_str, }; use rustfs_utils::{DEFAULT_SIP_HASH_KEY, sip_hash}; +#[cfg(test)] use s3s::dto::ReplicationConfiguration; -use serde::Deserialize; -use serde::Serialize; use std::collections::HashMap; use std::sync::Arc; use time::OffsetDateTime; @@ -83,10 +81,7 @@ use tokio_util::sync::CancellationToken; use tracing::{debug, error, instrument, trace, warn}; use uuid::Uuid; -const LOG_COMPONENT_ECSTORE: &str = "ecstore"; -const LOG_SUBSYSTEM_REPLICATION_RESYNC: &str = "replication_resync"; const EVENT_RESYNC_STATUS_UPDATE_SKIPPED: &str = "replication_resync_status_update_skipped"; -const EVENT_RESYNC_CONFIG_LOOKUP_SKIPPED: &str = "replication_resync_config_lookup_skipped"; const EVENT_RESYNC_OBJECT_PROCESSED: &str = "replication_resync_object_processed"; const EVENT_RESYNC_RUNTIME_SKIPPED: &str = "replication_resync_runtime_skipped"; const EVENT_REPLICATION_DELETE_SKIPPED: &str = "replication_delete_skipped"; @@ -955,285 +950,6 @@ pub(crate) async fn save_resync_status( Ok(()) } -async fn get_replication_config(bucket: &str) -> Result> { - ReplicationMetadataStore::optional_replication_config(bucket).await -} - -#[derive(Debug, Clone, Default, Serialize, Deserialize)] -pub struct ReplicationConfig { - pub config: Option, - pub remotes: Option, -} - -impl ReplicationConfig { - pub fn new(config: Option, remotes: Option) -> Self { - Self { config, remotes } - } - - pub fn is_empty(&self) -> bool { - self.config.is_none() - } - - pub fn replicate(&self, obj: &ObjectOpts) -> bool { - self.config.as_ref().is_some_and(|config| config.replicate(obj)) - } - - pub async fn resync( - &self, - oi: ObjectInfo, - dsc: ReplicateDecision, - status: &HashMap, - ) -> ResyncDecision { - if self.is_empty() { - return ResyncDecision::default(); - } - - let mut dsc = dsc; - - if oi.delete_marker { - let opts = ObjectOpts { - name: oi.name.clone(), - version_id: oi.version_id, - delete_marker: true, - op_type: ReplicationType::Delete, - existing_object: true, - ..Default::default() - }; - let arns = self - .config - .as_ref() - .map(|config| config.filter_target_arns(&opts)) - .unwrap_or_default(); - - if arns.is_empty() { - return ResyncDecision::default(); - } - - for arn in arns { - let mut opts = opts.clone(); - opts.target_arn = arn; - - dsc.set(ReplicateTargetDecision::new(opts.target_arn.clone(), self.replicate(&opts), false)); - } - - return self.resync_internal(oi, dsc, status); - } - - let mut user_defined = (*oi.user_defined).clone(); - user_defined.remove(AMZ_BUCKET_REPLICATION_STATUS); - - let dsc = must_replicate( - oi.bucket.as_str(), - &oi.name, - MustReplicateOptions::new(&user_defined, (*oi.user_tags).clone(), ReplicationType::ExistingObject, false), - ) - .await; - - self.resync_internal(oi, dsc, status) - } - - fn resync_internal( - &self, - oi: ObjectInfo, - dsc: ReplicateDecision, - status: &HashMap, - ) -> ResyncDecision { - let Some(remotes) = self.remotes.as_ref() else { - return ResyncDecision::default(); - }; - - if remotes.is_empty() { - return ResyncDecision::default(); - } - - let mut resync_decision = ResyncDecision::default(); - - for target in remotes.targets.iter() { - if let Some(decision) = dsc.targets_map.get(&target.arn) - && decision.replicate - { - resync_decision.targets.insert( - decision.arn.clone(), - resync_target_for_object( - &ReplicationResyncTargetObject { - mod_time: oi.mod_time, - user_defined: oi.user_defined.as_ref(), - }, - &target.arn, - &target.reset_id, - target.reset_before_date, - status.get(&decision.arn).unwrap_or(&ReplicationStatusType::Empty).clone(), - ), - ); - } - } - - resync_decision - } -} - -pub(crate) fn get_must_replicate_options( - user_defined: &HashMap, - user_tags: String, - _status: ReplicationStatusType, - op_type: ReplicationType, - opts: ObjectOptions, -) -> MustReplicateOptions { - MustReplicateOptions::new(user_defined, user_tags, op_type, opts.replication_request) -} - -/// Returns whether object version is a delete marker and if object qualifies for replication -pub(crate) async fn check_replicate_delete( - bucket: &str, - dobj: &ObjectToDelete, - oi: &ObjectInfo, - del_opts: &ObjectOptions, - gerr: Option, -) -> ReplicateDecision { - let rcfg = match get_replication_config(bucket).await { - Ok(Some(config)) => config, - Ok(None) => { - // warn!("No replication config found for bucket: {}", bucket); - return ReplicateDecision::default(); - } - Err(err) => { - error!( - event = EVENT_RESYNC_CONFIG_LOOKUP_SKIPPED, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, - bucket = %bucket, - reason = "replication_config_lookup_failed", - error = %err, - "Failed to look up replication config for delete replication" - ); - return ReplicateDecision::default(); - } - }; - - // If incoming request is a replication request, it does not need to be re-replicated. - if del_opts.replication_request { - return ReplicateDecision::default(); - } - - // Skip replication if this object's prefix is excluded from being versioned. - if !del_opts.versioned { - return ReplicateDecision::default(); - } - - let opts = delete_replication_object_opts( - dobj, - &ReplicationDeleteSource { - user_defined: oi.user_defined.as_ref(), - user_tags: oi.user_tags.as_str(), - delete_marker: oi.delete_marker, - replication_status: oi.replication_status.clone(), - }, - ); - - let tgt_arns = rcfg.filter_target_arns(&opts); - let mut dsc = ReplicateDecision::new(); - - if tgt_arns.is_empty() { - return dsc; - } - - for tgt_arn in tgt_arns { - let mut opts = opts.clone(); - opts.target_arn = tgt_arn.clone(); - let replicate = rcfg.replicate(&opts); - let sync = false; // Default sync value - - // When incoming delete is removal of a delete marker (a.k.a versioned delete), - // GetObjectInfo returns extra information even though it returns errFileNotFound - if gerr.is_some() { - if let Some(replicate) = delete_replication_missing_source_decision( - oi.delete_marker, - oi.target_replication_status(&tgt_arn), - replicate, - &oi.version_purge_status, - ) { - dsc.set(ReplicateTargetDecision::new(tgt_arn, replicate, sync)); - } - continue; - } - - let tgt = ReplicationTargetStore::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 { - ReplicateTargetDecision::new(tgt_arn, replicate, tgt.replicate_sync) - } else { - ReplicateTargetDecision::new(tgt_arn, false, false) - }; - dsc.set(tgt_dsc); - } - - dsc -} - -pub(crate) async fn must_replicate(bucket: &str, object: &str, mopts: MustReplicateOptions) -> ReplicateDecision { - if runtime_sources::object_store_handle().is_none() { - return ReplicateDecision::default(); - } - - if !ReplicationVersioningStore::prefix_enabled(bucket, object).await { - return ReplicateDecision::default(); - } - - let replication_status = mopts.replication_status(); - - if replication_status == ReplicationStatusType::Replica && !mopts.is_metadata_replication() { - return ReplicateDecision::default(); - } - - if mopts.is_replication_request() { - return ReplicateDecision::default(); - } - - let cfg = match get_replication_config(bucket).await { - Ok(cfg) => { - if let Some(cfg) = cfg { - cfg - } else { - return ReplicateDecision::default(); - } - } - Err(_err) => { - return ReplicateDecision::default(); - } - }; - - let opts = ObjectOpts { - name: object.to_string(), - replica: replication_status == ReplicationStatusType::Replica, - existing_object: mopts.is_existing_object_replication(), - user_tags: mopts.user_tags().to_string(), - ..Default::default() - }; - - let arns = cfg.filter_target_arns(&opts); - - if arns.is_empty() { - return ReplicateDecision::default(); - } - - let mut dsc = ReplicateDecision::default(); - - for arn in arns { - let cli = ReplicationTargetStore::remote_target_client(bucket, &arn).await; - - let mut sopts = opts.clone(); - sopts.target_arn = arn.clone(); - - let replicate = cfg.replicate(&sopts); - let synchronous = if let Some(cli) = cli { cli.replicate_sync } else { false }; - - dsc.set(ReplicateTargetDecision::new(arn, replicate, synchronous)); - } - - dsc -} - pub async fn replicate_delete(dobj: DeletedObjectReplicationInfo, storage: Arc) { if dobj.delete_object.force_delete { replicate_force_delete_to_targets(&dobj, storage).await; diff --git a/crates/ecstore/src/bucket/replication/replication_scanner_bridge.rs b/crates/ecstore/src/bucket/replication/replication_scanner_bridge.rs index 714929b8b..340a1bbd8 100644 --- a/crates/ecstore/src/bucket/replication/replication_scanner_bridge.rs +++ b/crates/ecstore/src/bucket/replication/replication_scanner_bridge.rs @@ -13,8 +13,8 @@ // limitations under the License. use super::ReplicationHealQueueResult; +use super::replication_object_config::ReplicationConfig; use super::replication_pool::queue_replication_heal_internal; -use super::replication_resyncer::ReplicationConfig; use super::replication_storage_boundary::ObjectInfo; pub struct ReplicationScannerBridge;