diff --git a/rustfs/src/admin/handlers/bucket_meta.rs b/rustfs/src/admin/handlers/bucket_meta.rs index 3cf50a50b..ebb72bbad 100644 --- a/rustfs/src/admin/handlers/bucket_meta.rs +++ b/rustfs/src/admin/handlers/bucket_meta.rs @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +use crate::admin::handlers::site_replication::site_replication_bucket_meta_hook; use crate::admin::storage_api::bucket::utils::{deserialize, serialize}; use crate::admin::storage_api::bucket::{ metadata::{ @@ -38,6 +39,7 @@ use http::{HeaderMap, StatusCode}; use hyper::Method; use matchit::Params; use rustfs_config::MAX_BUCKET_METADATA_IMPORT_SIZE; +use rustfs_madmin::{SITE_REPL_API_VERSION, SRBucketMeta}; use rustfs_policy::policy::{ BucketPolicy, action::{Action, AdminAction}, @@ -833,6 +835,7 @@ impl Operation for ImportBucketMetadata { // field and saves it, preserving any configs not present in the import archive. for (bucket_name, metadata) in &bucket_metadatas { for (config_file, data) in imported_configs_to_persist(metadata) { + let site_replication_item = imported_config_to_site_replication_item(bucket_name, metadata, config_file, &data)?; if let Err(e) = metadata_sys::update(bucket_name, config_file, data).await { warn!( event = EVENT_ADMIN_BUCKET_META_STATE, @@ -850,11 +853,24 @@ impl Operation for ImportBucketMetadata { "failed to persist imported bucket metadata for {bucket_name}/{config_file}: {e}" )); } + if let Some(item) = site_replication_item + && let Err(err) = site_replication_bucket_meta_hook(item).await + { + warn!( + event = EVENT_ADMIN_BUCKET_META_STATE, + component = LOG_COMPONENT_ADMIN, + subsystem = LOG_SUBSYSTEM_BUCKET_META, + action = "import_bucket_metadata", + result = "site_replication_notify_failed", + bucket = %bucket_name, + config_name = %config_file, + error = %err, + "admin bucket meta state" + ); + } } } - // TODO: site replication notify - let mut header = HeaderMap::new(); header.insert(CONTENT_TYPE, "application/json".parse().expect("valid header value")); header.insert(CONTENT_LENGTH, "0".parse().expect("valid header value")); @@ -887,6 +903,80 @@ fn imported_configs_to_persist(metadata: &BucketMetadata) -> Vec<(&'static str, .collect() } +fn imported_config_to_site_replication_item( + bucket_name: &str, + metadata: &BucketMetadata, + config_file: &str, + data: &[u8], +) -> S3Result> { + let mut item = match config_file { + BUCKET_POLICY_CONFIG => site_replication_bucket_meta_item(bucket_name, "policy", metadata.policy_config_updated_at), + BUCKET_LIFECYCLE_CONFIG => { + site_replication_bucket_meta_item(bucket_name, "lc-config", metadata.lifecycle_config_updated_at) + } + BUCKET_SSECONFIG => site_replication_bucket_meta_item(bucket_name, "sse-config", metadata.encryption_config_updated_at), + BUCKET_TAGGING_CONFIG => site_replication_bucket_meta_item(bucket_name, "tags", metadata.tagging_config_updated_at), + BUCKET_QUOTA_CONFIG_FILE => { + site_replication_bucket_meta_item(bucket_name, "quota-config", metadata.quota_config_updated_at) + } + OBJECT_LOCK_CONFIG => { + site_replication_bucket_meta_item(bucket_name, "object-lock-config", metadata.object_lock_config_updated_at) + } + BUCKET_VERSIONING_CONFIG => { + site_replication_bucket_meta_item(bucket_name, "version-config", metadata.versioning_config_updated_at) + } + BUCKET_REPLICATION_CONFIG => { + site_replication_bucket_meta_item(bucket_name, "replication-config", metadata.replication_config_updated_at) + } + _ => return Ok(None), + }; + + match config_file { + BUCKET_POLICY_CONFIG => { + item.policy = + Some(serde_json::from_slice(data).map_err(|e| { + s3_error!(InternalError, "failed to encode imported bucket policy for site replication: {e}") + })?); + } + BUCKET_LIFECYCLE_CONFIG => { + set_imported_config_string(&mut item.expiry_lc_config, config_file, data)?; + item.expiry_updated_at = item.updated_at; + } + BUCKET_SSECONFIG => set_imported_config_string(&mut item.sse_config, config_file, data)?, + BUCKET_TAGGING_CONFIG => set_imported_config_string(&mut item.tags, config_file, data)?, + BUCKET_QUOTA_CONFIG_FILE => { + item.quota = Some( + serde_json::from_slice(data) + .map_err(|e| s3_error!(InternalError, "failed to encode imported bucket quota for site replication: {e}"))?, + ); + } + OBJECT_LOCK_CONFIG => set_imported_config_string(&mut item.object_lock_config, config_file, data)?, + BUCKET_VERSIONING_CONFIG => set_imported_config_string(&mut item.versioning, config_file, data)?, + BUCKET_REPLICATION_CONFIG => set_imported_config_string(&mut item.replication_config, config_file, data)?, + _ => unreachable!(), + } + + Ok(Some(item)) +} + +fn site_replication_bucket_meta_item(bucket_name: &str, item_type: &str, updated_at: OffsetDateTime) -> SRBucketMeta { + SRBucketMeta { + bucket: bucket_name.to_string(), + r#type: item_type.to_string(), + updated_at: Some(updated_at), + api_version: Some(SITE_REPL_API_VERSION.to_string()), + ..Default::default() + } +} + +fn set_imported_config_string(target: &mut Option, config_file: &str, data: &[u8]) -> S3Result<()> { + *target = Some( + String::from_utf8(data.to_vec()) + .map_err(|e| s3_error!(InternalError, "imported bucket metadata {config_file} is not valid UTF-8: {e}"))?, + ); + Ok(()) +} + #[cfg(test)] mod import_persist_tests { use super::*; @@ -919,4 +1009,64 @@ mod import_persist_tests { let metadata = BucketMetadata::new("untouched-bucket"); assert!(imported_configs_to_persist(&metadata).is_empty()); } + + #[test] + fn imported_configs_are_scheduled_for_site_replication() { + let updated_at = OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(42); + let mut metadata = BucketMetadata::new("restored-bucket"); + metadata.policy_config_json = br#"{"Version":"2012-10-17","Statement":[]}"#.to_vec(); + metadata.policy_config_updated_at = updated_at; + metadata.versioning_config_xml = b"Enabled".to_vec(); + metadata.versioning_config_updated_at = updated_at; + metadata.lifecycle_config_xml = b"".to_vec(); + metadata.lifecycle_config_updated_at = updated_at; + metadata.quota_config_json = br#"{"quota":1024}"#.to_vec(); + metadata.quota_config_updated_at = updated_at; + + let mut items = imported_configs_to_persist(&metadata) + .into_iter() + .filter_map(|(config_file, data)| { + imported_config_to_site_replication_item("restored-bucket", &metadata, config_file, &data) + .expect("site replication item should encode") + }) + .collect::>(); + items.sort_by(|left, right| left.r#type.cmp(&right.r#type)); + + assert_eq!(items.len(), 4); + assert!(items.iter().all(|item| item.bucket == "restored-bucket")); + assert!(items.iter().all(|item| item.updated_at == Some(updated_at))); + assert!( + items + .iter() + .all(|item| item.api_version.as_deref() == Some(SITE_REPL_API_VERSION)) + ); + assert!(items.iter().any(|item| item.r#type == "policy" && item.policy.is_some())); + assert!( + items + .iter() + .any(|item| item.r#type == "version-config" && item.versioning.is_some()) + ); + assert!(items.iter().any(|item| { + item.r#type == "lc-config" && item.expiry_lc_config.is_some() && item.expiry_updated_at == Some(updated_at) + })); + assert!(items.iter().any(|item| item.r#type == "quota-config" && item.quota.is_some())); + } + + #[test] + fn unsupported_imported_configs_are_not_site_replicated() { + let mut metadata = BucketMetadata::new("restored-bucket"); + metadata.notification_config_xml = b"".to_vec(); + metadata.bucket_targets_config_json = b"[]".to_vec(); + + let has_site_replication_item = imported_configs_to_persist(&metadata) + .into_iter() + .filter_map(|(config_file, data)| { + imported_config_to_site_replication_item("restored-bucket", &metadata, config_file, &data) + .expect("unsupported configs should be skipped") + }) + .next() + .is_some(); + + assert!(!has_site_replication_item); + } }