From 45b675c64140e484c5ee806dbce6a0abafbaf377 Mon Sep 17 00:00:00 2001 From: cxymds Date: Sat, 25 Jul 2026 01:32:50 +0800 Subject: [PATCH] fix(replication): report authoritative backlog metrics (#5209) --- crates/ecstore/src/api/mod.rs | 25 +- crates/ecstore/src/bucket/replication/mod.rs | 10 +- .../replication_filemeta_boundary.rs | 2 +- .../bucket/replication/replication_pool.rs | 76 +++++ .../bucket/replication/replication_state.rs | 239 ++++++++++++-- .../replication/replication_stats_boundary.rs | 4 +- .../src/cluster/rpc/peer_rest_client.rs | 105 ++++++- crates/replication/src/lib.rs | 2 +- crates/replication/src/stats.rs | 94 +++++- rustfs/src/admin/handlers/replication.rs | 293 ++++++++++++++---- rustfs/src/admin/router.rs | 55 +++- rustfs/src/admin/runtime_sources.rs | 4 +- rustfs/src/admin/storage_api.rs | 9 + rustfs/src/app/context.rs | 2 +- rustfs/src/runtime_sources.rs | 8 +- rustfs/src/storage/rpc/node_service.rs | 34 +- rustfs/src/storage/runtime_sources.rs | 6 + rustfs/src/storage/storage_api.rs | 11 + 18 files changed, 839 insertions(+), 140 deletions(-) diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index f16de641a..97c5bee7e 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -149,18 +149,19 @@ pub mod bucket { pub mod replication { pub use crate::bucket::replication::{ - BucketReplicationResyncStatus, BucketStats, DeletedObjectReplicationInfo, DynReplicationPool, MustReplicateOptions, - ObjectOpts, REPLICATE_INCOMING_DELETE, ReplicateDecision, ReplicateObjectInfo, ReplicationConfig, - ReplicationConfigurationExt, ReplicationDeleteScheduleInput, ReplicationDeleteStateSource, - ReplicationHealQueueResult, ReplicationObjectBridge, ReplicationObjectIO, ReplicationOperation, ReplicationPoolTrait, - ReplicationPriority, ReplicationQueueAdmission, ReplicationScannerBridge, ReplicationState, ReplicationStats, - ReplicationStatusType, ReplicationStorage, ReplicationTargetValidationError, ReplicationType, ResyncOpts, - ResyncStatusType, TargetReplicationResyncStatus, VersionPurgeStatusType, delete_replication_state_from_config, - delete_replication_version_id, get_global_replication_pool, get_global_replication_stats, - init_background_replication, replication_state_to_filemeta, replication_status_to_filemeta, replication_statuses_map, - replication_target_arns, should_remove_replication_target, should_schedule_delete_replication, - should_use_existing_delete_replication_info, should_use_existing_delete_replication_source, - validate_replication_config_target_arns, version_purge_status_to_filemeta, + BucketReplicationResyncStatus, BucketStats, DeletedObjectReplicationInfo, DurableMrfBacklog, DynReplicationPool, + MrfOpKind, MrfReplicateEntry, MustReplicateOptions, ObjectOpts, REPLICATE_INCOMING_DELETE, ReplicateDecision, + ReplicateObjectInfo, ReplicationConfig, ReplicationConfigurationExt, ReplicationDeleteScheduleInput, + ReplicationDeleteStateSource, ReplicationHealQueueResult, ReplicationObjectBridge, ReplicationObjectIO, + ReplicationOperation, ReplicationPoolTrait, ReplicationPriority, ReplicationQueueAdmission, ReplicationScannerBridge, + ReplicationState, ReplicationStats, ReplicationStatusType, ReplicationStorage, ReplicationTargetValidationError, + ReplicationType, ResyncOpts, ResyncStatusType, TargetReplicationResyncStatus, VersionPurgeStatusType, + delete_replication_state_from_config, delete_replication_version_id, get_global_replication_pool, + get_global_replication_stats, init_background_replication, read_durable_mrf_backlog, replication_state_to_filemeta, + replication_status_to_filemeta, replication_statuses_map, replication_target_arns, should_remove_replication_target, + should_schedule_delete_replication, should_use_existing_delete_replication_info, + should_use_existing_delete_replication_source, validate_replication_config_target_arns, + version_purge_status_to_filemeta, }; } diff --git a/crates/ecstore/src/bucket/replication/mod.rs b/crates/ecstore/src/bucket/replication/mod.rs index 5e1aa0a28..eb04a752c 100644 --- a/crates/ecstore/src/bucket/replication/mod.rs +++ b/crates/ecstore/src/bucket/replication/mod.rs @@ -52,9 +52,9 @@ pub use replication_config_boundary::{ pub(crate) use replication_filemeta_boundary::ReplicateTargetDecision; pub(crate) use replication_filemeta_boundary::version_purge_statuses_map; pub use replication_filemeta_boundary::{ - REPLICATE_INCOMING_DELETE, ReplicateDecision, ReplicateObjectInfo, ReplicationState, ReplicationStatusType, ReplicationType, - VersionPurgeStatusType, replication_state_to_filemeta, replication_status_to_filemeta, replication_statuses_map, - version_purge_status_to_filemeta, + MrfOpKind, MrfReplicateEntry, REPLICATE_INCOMING_DELETE, ReplicateDecision, ReplicateObjectInfo, ReplicationState, + ReplicationStatusType, ReplicationType, VersionPurgeStatusType, replication_state_to_filemeta, + replication_status_to_filemeta, replication_statuses_map, version_purge_status_to_filemeta, }; pub(crate) use replication_filemeta_boundary::{ replication_state_from_filemeta, replication_status_from_filemeta, version_purge_status_from_filemeta, @@ -69,8 +69,8 @@ pub use replication_object_decision_boundary::{ should_use_existing_delete_replication_source, }; pub use replication_pool::{ - DynReplicationPool, ReplicationPoolTrait, get_global_replication_pool, get_global_replication_stats, - init_background_replication, + DurableMrfBacklog, DynReplicationPool, ReplicationPoolTrait, get_global_replication_pool, get_global_replication_stats, + init_background_replication, read_durable_mrf_backlog, }; pub use replication_queue_boundary::{ DeletedObjectReplicationInfo, ReplicationHealQueueResult, ReplicationOperation, ReplicationPriority, diff --git a/crates/ecstore/src/bucket/replication/replication_filemeta_boundary.rs b/crates/ecstore/src/bucket/replication/replication_filemeta_boundary.rs index f394921be..536f89717 100644 --- a/crates/ecstore/src/bucket/replication/replication_filemeta_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_filemeta_boundary.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -pub(crate) use rustfs_replication::{MrfOpKind, MrfReplicateEntry}; +pub use rustfs_replication::{MrfOpKind, MrfReplicateEntry}; pub(crate) use rustfs_replication::{ REPLICATE_EXISTING, REPLICATE_EXISTING_DELETE, REPLICATE_HEAL_DELETE, ReplicateTargetDecision, ReplicatedInfos, ReplicatedTargetInfo, ReplicationAction, ReplicationWorkerOperation, ResyncDecision, get_replication_state, diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs index 5b279eb7e..c25e49d6e 100644 --- a/crates/ecstore/src/bucket/replication/replication_pool.rs +++ b/crates/ecstore/src/bucket/replication/replication_pool.rs @@ -68,6 +68,33 @@ const EVENT_REPLICATION_RESYNC_LOAD_SKIPPED: &str = "replication_resync_load_ski const EVENT_REPLICATION_RESYNC_RECOVERED: &str = "replication_resync_recovered"; const EVENT_REPLICATION_MRF_QUEUE_UNAVAILABLE: &str = "replication_mrf_queue_unavailable"; +#[derive(Debug, Default)] +pub struct DurableMrfBacklog { + pub available: bool, + pub entries: Vec, +} + +fn durable_mrf_backlog_from_read(result: Result, EcstoreError>) -> DurableMrfBacklog { + match result { + Ok(data) => match decode_mrf_file(&data) { + Ok(entries) if entries.iter().all(|entry| entry.size >= 0) => DurableMrfBacklog { + available: true, + entries, + }, + Ok(_) | Err(_) => DurableMrfBacklog::default(), + }, + Err(EcstoreError::ConfigNotFound) => DurableMrfBacklog { + available: true, + entries: Vec::new(), + }, + Err(_) => DurableMrfBacklog::default(), + } +} + +pub async fn read_durable_mrf_backlog(storage: Arc) -> DurableMrfBacklog { + durable_mrf_backlog_from_read(ReplicationConfigStore::read(storage, ReplicationMetadataStore::MRF_REPLICATION_FILE).await) +} + /// Main replication pool structure #[derive(Debug)] pub struct ReplicationPool { @@ -2233,4 +2260,53 @@ mod tests { // None so replay falls back to the current time (backlog#867 backward compatibility). assert_eq!(entry.delete_marker_mtime, None, "missing deleteMarkerMtime key must default to None"); } + + #[test] + fn durable_mrf_snapshot_reads_restart_backlog_and_valid_empty_state() { + let entries = vec![MrfReplicateEntry { + bucket: "restart-bucket".to_string(), + object: "object".to_string(), + version_id: None, + retry_count: 1, + size: 512, + op: MrfOpKind::Object, + delete_marker_version_id: None, + delete_marker: false, + delete_marker_mtime: None, + }]; + let encoded = encode_mrf_file(&entries).expect("durable MRF backlog should encode"); + + let recovered = durable_mrf_backlog_from_read(Ok(encoded)); + assert!(recovered.available); + assert_eq!(recovered.entries.len(), 1); + assert_eq!(recovered.entries[0].bucket, "restart-bucket"); + assert_eq!(recovered.entries[0].size, 512); + + let missing_file = durable_mrf_backlog_from_read(Err(EcstoreError::ConfigNotFound)); + assert!(missing_file.available); + assert!(missing_file.entries.is_empty()); + } + + #[test] + fn durable_mrf_snapshot_marks_corrupt_or_invalid_data_unavailable() { + let corrupt = durable_mrf_backlog_from_read(Ok(vec![0, 1, 2])); + assert!(!corrupt.available); + assert!(corrupt.entries.is_empty()); + + let negative = encode_mrf_file(&[MrfReplicateEntry { + bucket: "bucket".to_string(), + object: "object".to_string(), + version_id: None, + retry_count: 0, + size: -1, + op: MrfOpKind::Object, + delete_marker_version_id: None, + delete_marker: false, + delete_marker_mtime: None, + }]) + .expect("invalid persisted entry should still encode for boundary testing"); + let invalid = durable_mrf_backlog_from_read(Ok(negative)); + assert!(!invalid.available); + assert!(invalid.entries.is_empty()); + } } diff --git a/crates/ecstore/src/bucket/replication/replication_state.rs b/crates/ecstore/src/bucket/replication/replication_state.rs index 7b54d474d..11a5e99e7 100644 --- a/crates/ecstore/src/bucket/replication/replication_state.rs +++ b/crates/ecstore/src/bucket/replication/replication_state.rs @@ -15,9 +15,11 @@ use super::replication_error_boundary::Error; use super::replication_filemeta_boundary::{ReplicatedTargetInfo, ReplicationStatusType, ReplicationType}; use super::replication_resync_boundary::ResyncStatusType; +#[cfg(test)] +use super::replication_stats_boundary::FailStats; use super::replication_stats_boundary::{ ActiveWorkerStat, BucketReplicationStat, BucketReplicationStats, BucketStats, InQueueMetric, ProxyMetric, ProxyStatsCache, - QueueCache, SRMetricsSummary, XferStats, + QueueCache, ReplicationMetricScope, SRMetricsSummary, XferStats, }; use super::runtime_boundary as runtime_sources; use std::collections::HashMap; @@ -361,10 +363,12 @@ impl ReplicationStats { if rs.transfer_duration > Duration::default() { stat.latency.update(rs.transfer_size, rs.transfer_duration); stat.update_xfer_rate(rs.transfer_size, rs.transfer_duration); + stat.latency_scope = ReplicationMetricScope::NodeLocal; } } (false, true, false) => { stat.fail_stats.add_size(rs.transfer_size, rs.err.as_ref()); + stat.failed = stat.fail_stats.to_metric(); } (false, false, true) => { // Pending status, no processing for now @@ -379,7 +383,10 @@ impl ReplicationStats { let mut result = HashMap::with_capacity(cache.len()); for (bucket, stats) in cache.iter() { - result.insert(bucket.clone(), stats.clone_stats()); + let mut snapshot = stats.clone_stats(); + snapshot.mark_node_local_provider_available(); + snapshot.queue_scope = ReplicationMetricScope::NodeLocal; + result.insert(bucket.clone(), snapshot); } drop(cache); @@ -388,6 +395,8 @@ impl ReplicationStats { for (bucket, queue_stats) in &q_cache.bucket_stats { let bucket_stats = result.entry(bucket.clone()).or_insert_with(BucketReplicationStats::new); bucket_stats.q_stat = queue_stats.snapshot(); + bucket_stats.mark_node_local_provider_available(); + bucket_stats.queue_scope = ReplicationMetricScope::NodeLocal; } } @@ -405,9 +414,13 @@ impl ReplicationStats { pub async fn get(&self, bucket: &str) -> BucketReplicationStats { let cache = self.cache.read().await; if let Some(stats) = cache.get(bucket) { - stats.clone_stats() + let mut snapshot = stats.clone_stats(); + snapshot.mark_node_local_provider_available(); + snapshot } else { - BucketReplicationStats::new() + let mut snapshot = BucketReplicationStats::new(); + snapshot.mark_node_local_provider_available(); + snapshot } } @@ -453,11 +466,15 @@ impl ReplicationStats { let mut tq = InQueueMetric::default(); for bucket_stat in &bucket_stats { - tot_replica_size += bucket_stat.replication_stats.replica_size; - tot_replica_count += bucket_stat.replication_stats.replica_count; + tot_replica_size = tot_replica_size.saturating_add(bucket_stat.replication_stats.replica_size); + tot_replica_count = tot_replica_count.saturating_add(bucket_stat.replication_stats.replica_count); - for q in &bucket_stat.queue_stats.nodes { - tq = tq.merge(&q.q_stats); + if bucket_stat.replication_stats.queue_scope != ReplicationMetricScope::Unavailable { + tq = tq.merge(&bucket_stat.replication_stats.q_stat); + } else { + for q in &bucket_stat.queue_stats.nodes { + tq = tq.merge(&q.q_stats); + } } for (arn, stat) in &bucket_stat.replication_stats.stats { @@ -470,22 +487,38 @@ impl ReplicationStats { let f_stats = stat.fail_stats.merge(&old_stat.fail_stats); let lrg = old_stat.xfer_rate_lrg.merge(&stat.xfer_rate_lrg); let sml = old_stat.xfer_rate_sml.merge(&stat.xfer_rate_sml); + let latency_available = stat.latency_scope != ReplicationMetricScope::Unavailable + || old_stat.latency_scope != ReplicationMetricScope::Unavailable; + let bandwidth_available = stat.bandwidth_scope != ReplicationMetricScope::Unavailable + || old_stat.bandwidth_scope != ReplicationMetricScope::Unavailable; *old_stat = BucketReplicationStat { failed: f_stats.to_metric(), fail_stats: f_stats, - replicated_size: stat.replicated_size + old_stat.replicated_size, - replicated_count: stat.replicated_count + old_stat.replicated_count, + replicated_size: stat.replicated_size.saturating_add(old_stat.replicated_size), + replicated_count: stat.replicated_count.saturating_add(old_stat.replicated_count), latency: stat.latency.merge(&old_stat.latency), xfer_rate_lrg: lrg, xfer_rate_sml: sml, - bandwidth_limit_bytes_per_sec: stat.bandwidth_limit_bytes_per_sec, + bandwidth_limit_bytes_per_sec: stat + .bandwidth_limit_bytes_per_sec + .saturating_add(old_stat.bandwidth_limit_bytes_per_sec), current_bandwidth_bytes_per_sec: stat.current_bandwidth_bytes_per_sec + old_stat.current_bandwidth_bytes_per_sec, + latency_scope: if latency_available { + ReplicationMetricScope::ClusterAggregated + } else { + ReplicationMetricScope::Unavailable + }, + bandwidth_scope: if bandwidth_available { + ReplicationMetricScope::ClusterAggregated + } else { + ReplicationMetricScope::Unavailable + }, }; - tot_replicated_size += stat.replicated_size; - tot_replicated_count += stat.replicated_count; + tot_replicated_size = tot_replicated_size.saturating_add(stat.replicated_size); + tot_replicated_count = tot_replicated_count.saturating_add(stat.replicated_count); } } @@ -499,23 +532,28 @@ impl ReplicationStats { resync_started_count: bucket_stats .iter() .map(|stats| stats.replication_stats.resync_started_count) - .sum(), + .fold(0i64, i64::saturating_add), resync_completed_count: bucket_stats .iter() .map(|stats| stats.replication_stats.resync_completed_count) - .sum(), + .fold(0i64, i64::saturating_add), resync_failed_count: bucket_stats .iter() .map(|stats| stats.replication_stats.resync_failed_count) - .sum(), + .fold(0i64, i64::saturating_add), resync_canceled_count: bucket_stats .iter() .map(|stats| stats.replication_stats.resync_canceled_count) - .sum(), + .fold(0i64, i64::saturating_add), resync_duration_ms: bucket_stats .iter() .map(|stats| stats.replication_stats.resync_duration_ms) - .sum(), + .fold(0i64, i64::saturating_add), + provider_available: true, + cluster_complete: true, + observed_node_count: u32::try_from(bucket_stats.len()).unwrap_or(u32::MAX), + expected_node_count: u32::try_from(bucket_stats.len()).unwrap_or(u32::MAX), + queue_scope: ReplicationMetricScope::ClusterAggregated, }; let qs = Default::default(); @@ -547,6 +585,33 @@ impl ReplicationStats { bs } + pub async fn aggregate_bucket_replication_stats( + &self, + bucket: &str, + bucket_stats: Vec, + expected_node_count: u32, + ) -> BucketStats { + let mut aggregated = self.calculate_bucket_replication_stats(bucket, bucket_stats).await; + let observed_node_count = aggregated.replication_stats.observed_node_count; + let complete = observed_node_count == expected_node_count; + aggregated.replication_stats.expected_node_count = expected_node_count; + aggregated.replication_stats.cluster_complete = complete; + aggregated.replication_stats.queue_scope = if complete { + ReplicationMetricScope::ClusterAggregated + } else { + ReplicationMetricScope::PartialCluster + }; + for stat in aggregated.replication_stats.stats.values_mut() { + if stat.latency_scope != ReplicationMetricScope::Unavailable { + stat.latency_scope = aggregated.replication_stats.queue_scope; + } + if stat.bandwidth_scope != ReplicationMetricScope::Unavailable { + stat.bandwidth_scope = aggregated.replication_stats.queue_scope; + } + } + aggregated + } + /// Get latest replication statistics pub async fn get_latest_replication_stats(&self, bucket: &str) -> BucketStats { // In actual implementation, statistics would be obtained from cluster @@ -567,6 +632,15 @@ impl ReplicationStats { }; drop(cache); + { + let q_cache = self.q_cache.lock().await; + if let Some(queue_stats) = q_cache.bucket_stats.get(bucket) { + replication_stats.q_stat = queue_stats.snapshot(); + } + } + replication_stats.mark_node_local_provider_available(); + replication_stats.queue_scope = ReplicationMetricScope::NodeLocal; + if let Some(monitor) = runtime_sources::bucket_monitor() { let bw_report = monitor.get_report(|name| name == bucket); for (opts, bw) in bw_report.bucket_stats { @@ -578,8 +652,7 @@ impl ReplicationStats { xfer_rate_sml: XferStats::new(), ..Default::default() }); - stat.bandwidth_limit_bytes_per_sec = bw.limit_bytes_per_sec; - stat.current_bandwidth_bytes_per_sec = bw.current_bandwidth_bytes_per_sec; + stat.set_node_local_bandwidth(bw.limit_bytes_per_sec, bw.current_bandwidth_bytes_per_sec); } } @@ -724,6 +797,132 @@ mod tests { assert_eq!(stat.replicated_count, 1); } + #[tokio::test] + async fn latest_stats_include_queue_until_drained() { + let stats = ReplicationStats::new(); + + stats.inc_q("queued-bucket", 4096, false, ReplicationType::Object).await; + let queued = stats.get_latest_replication_stats("queued-bucket").await; + assert!(queued.replication_stats.provider_available); + assert_eq!(queued.replication_stats.q_stat.curr.count, 1); + assert_eq!(queued.replication_stats.q_stat.curr.bytes, 4096); + assert_eq!(queued.replication_stats.queue_scope, ReplicationMetricScope::NodeLocal); + + stats.dec_q("queued-bucket", 4096, false, ReplicationType::Object).await; + let drained = stats.get_latest_replication_stats("queued-bucket").await; + assert_eq!(drained.replication_stats.q_stat.curr.count, 0); + assert_eq!(drained.replication_stats.q_stat.curr.bytes, 0); + } + + #[tokio::test] + async fn failed_metric_matches_authoritative_fail_stats() { + let stats = ReplicationStats::new(); + let target_info = ReplicatedTargetInfo { + arn: "failed-arn".to_string(), + size: 2048, + duration: Duration::from_millis(25), + op_type: ReplicationType::Object, + error: Some("target unavailable".to_string()), + ..Default::default() + }; + + stats + .update( + "failed-bucket", + &target_info, + ReplicationStatusType::Failed, + ReplicationStatusType::Pending, + ) + .await; + + let snapshot = stats.get_latest_replication_stats("failed-bucket").await; + let target = &snapshot.replication_stats.stats["failed-arn"]; + assert_eq!(target.failed.count, target.fail_stats.count); + assert_eq!(target.failed.size, target.fail_stats.size); + assert_eq!(target.failed.count, 1); + assert_eq!(target.failed.size, 2048); + } + + #[tokio::test] + async fn valid_empty_provider_is_not_reported_as_unavailable() { + let stats = ReplicationStats::new(); + + let snapshot = stats.get_latest_replication_stats("empty-bucket").await; + + assert!(snapshot.replication_stats.provider_available); + assert!(snapshot.replication_stats.cluster_complete); + assert_eq!(snapshot.replication_stats.observed_node_count, 1); + assert_eq!(snapshot.replication_stats.expected_node_count, 1); + assert!(snapshot.replication_stats.stats.is_empty()); + } + + #[tokio::test] + async fn cluster_aggregation_counts_each_node_once_and_marks_partial() { + let stats = ReplicationStats::new(); + let node = |failed_count, failed_size, queued_count, queued_size| { + let mut fail_stats = FailStats::new(); + fail_stats.count = failed_count; + fail_stats.size = failed_size; + let mut targets = HashMap::new(); + targets.insert( + "arn".to_string(), + BucketReplicationStat { + fail_stats, + latency_scope: ReplicationMetricScope::NodeLocal, + ..Default::default() + }, + ); + let q_stat = InQueueMetric::default(); + q_stat.curr.now_count.store(queued_count, Ordering::Relaxed); + q_stat.curr.now_bytes.store(queued_size, Ordering::Relaxed); + let q_stat = q_stat.snapshot(); + BucketStats { + replication_stats: BucketReplicationStats { + stats: targets, + q_stat, + provider_available: true, + queue_scope: ReplicationMetricScope::NodeLocal, + ..Default::default() + }, + ..Default::default() + } + }; + + let aggregated = stats + .aggregate_bucket_replication_stats("bucket", vec![node(1, 10, 2, 20), node(3, 30, 4, 40)], 3) + .await; + + let target = &aggregated.replication_stats.stats["arn"]; + assert_eq!(target.failed.count, 4); + assert_eq!(target.failed.size, 40); + assert_eq!(aggregated.replication_stats.q_stat.curr.count, 6); + assert_eq!(aggregated.replication_stats.q_stat.curr.bytes, 60); + assert_eq!(aggregated.replication_stats.observed_node_count, 2); + assert_eq!(aggregated.replication_stats.expected_node_count, 3); + assert!(!aggregated.replication_stats.cluster_complete); + assert_eq!(aggregated.replication_stats.queue_scope, ReplicationMetricScope::PartialCluster); + assert_eq!(target.latency_scope, ReplicationMetricScope::PartialCluster); + } + + #[tokio::test] + async fn concurrent_queue_updates_are_visible_without_lost_counts() { + let stats = Arc::new(ReplicationStats::new()); + let mut tasks = Vec::with_capacity(32); + for _ in 0..32 { + let stats = Arc::clone(&stats); + tasks.push(tokio::spawn(async move { + stats.inc_q("concurrent-bucket", 7, false, ReplicationType::Object).await; + })); + } + for task in tasks { + task.await.expect("queue update task should complete"); + } + + let snapshot = stats.get_latest_replication_stats("concurrent-bucket").await; + assert_eq!(snapshot.replication_stats.q_stat.curr.count, 32); + assert_eq!(snapshot.replication_stats.q_stat.curr.bytes, 224); + } + #[tokio::test] async fn test_get_all_includes_proxy_only_bucket() { let stats = ReplicationStats::new(); diff --git a/crates/ecstore/src/bucket/replication/replication_stats_boundary.rs b/crates/ecstore/src/bucket/replication/replication_stats_boundary.rs index 4f12638f4..e5bdfc66a 100644 --- a/crates/ecstore/src/bucket/replication/replication_stats_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_stats_boundary.rs @@ -13,7 +13,9 @@ // limitations under the License. pub use rustfs_replication::BucketStats; +#[cfg(test)] +pub(crate) use rustfs_replication::FailStats; pub(crate) use rustfs_replication::{ ActiveWorkerStat, BucketReplicationStat, BucketReplicationStats, InQueueMetric, ProxyMetric, ProxyStatsCache, QueueCache, - SRMetricsSummary, XferStats, + ReplicationMetricScope, SRMetricsSummary, XferStats, }; diff --git a/crates/ecstore/src/cluster/rpc/peer_rest_client.rs b/crates/ecstore/src/cluster/rpc/peer_rest_client.rs index 0ce25d848..7bdda0d6d 100644 --- a/crates/ecstore/src/cluster/rpc/peer_rest_client.rs +++ b/crates/ecstore/src/cluster/rpc/peer_rest_client.rs @@ -19,6 +19,7 @@ use crate::cluster::rpc::client::{ use crate::cluster::rpc::{set_tonic_canonical_body_digest, verify_tonic_rpc_response_proof}; use crate::error::{Error, Result}; use crate::{ + bucket::replication::BucketStats, disk::disk_store::{get_drive_active_check_interval, get_drive_active_check_timeout}, layout::endpoints::EndpointServerPools, runtime::sources as runtime_sources, @@ -34,14 +35,15 @@ use rustfs_madmin::{ }; use rustfs_protos::proto_gen::node_service::{ BackgroundHealStatusRequest, CancelDecommissionRequest, ClearDecommissionRequest, DeleteBucketMetadataRequest, - DeletePolicyRequest, DeleteServiceAccountRequest, DeleteUserRequest, GetCpusRequest, GetLiveEventsRequest, GetMemInfoRequest, - GetMetricsRequest, GetNetInfoRequest, GetOsInfoRequest, GetPartitionsRequest, GetProcInfoRequest, GetSeLinuxInfoRequest, - GetSysConfigRequest, GetSysErrorsRequest, HealControlRequest, LoadBucketMetadataRequest, LoadGroupRequest, - LoadPolicyMappingRequest, LoadPolicyRequest, LoadRebalanceMetaRequest, LoadServiceAccountRequest, - LoadTransitionTierConfigRequest, LoadUserRequest, LocalStorageInfoRequest, Mss, ReloadPoolMetaRequest, - ReloadSiteReplicationConfigRequest, ScannerActivityRequest, ScannerActivityResponse, ServerInfoRequest, SignalServiceRequest, - StartDecommissionRequest, StartProfilingRequest, StopRebalanceRequest, TierMutationAbortRequest, TierMutationCommitRequest, - TierMutationControlResponse, TierMutationPeerState, TierMutationPrepareRequest, node_service_client::NodeServiceClient, + DeletePolicyRequest, DeleteServiceAccountRequest, DeleteUserRequest, GetBucketStatsDataRequest, GetBucketStatsDataResponse, + GetCpusRequest, GetLiveEventsRequest, GetMemInfoRequest, GetMetricsRequest, GetNetInfoRequest, GetOsInfoRequest, + GetPartitionsRequest, GetProcInfoRequest, GetSeLinuxInfoRequest, GetSysConfigRequest, GetSysErrorsRequest, + HealControlRequest, LoadBucketMetadataRequest, LoadGroupRequest, LoadPolicyMappingRequest, LoadPolicyRequest, + LoadRebalanceMetaRequest, LoadServiceAccountRequest, LoadTransitionTierConfigRequest, LoadUserRequest, + LocalStorageInfoRequest, Mss, ReloadPoolMetaRequest, ReloadSiteReplicationConfigRequest, ScannerActivityRequest, + ScannerActivityResponse, ServerInfoRequest, SignalServiceRequest, StartDecommissionRequest, StartProfilingRequest, + StopRebalanceRequest, TierMutationAbortRequest, TierMutationCommitRequest, TierMutationControlResponse, + TierMutationPeerState, TierMutationPrepareRequest, node_service_client::NodeServiceClient, tier_mutation_control_service_client::TierMutationControlServiceClient, }; use rustfs_protos::{TierMutationRpcPhase, evict_failed_connection}; @@ -74,6 +76,26 @@ const HEAL_CONTROL_PAYLOAD_MAX_SIZE: usize = 64 * 1024; const PEER_REST_RECOVERY_MAX_ATTEMPTS: u32 = 60; const PEER_REST_RECOVERY_MAX_BACKOFF: Duration = Duration::from_secs(30); const SCANNER_ACTIVITY_MAX_MESSAGE_SIZE: usize = 1024; +const REPLICATION_STATS_MAX_MESSAGE_SIZE: usize = 8 * 1024 * 1024; + +fn decode_bucket_stats_response(response: GetBucketStatsDataResponse) -> Result { + if !response.success { + return Err(Error::other( + response + .error_info + .unwrap_or_else(|| "peer replication statistics provider is unavailable".to_string()), + )); + } + if response.bucket_stats.len() > REPLICATION_STATS_MAX_MESSAGE_SIZE { + return Err(Error::other("peer replication statistics response exceeds size limit")); + } + let mut buf = Deserializer::new(Cursor::new(response.bucket_stats)); + let stats = BucketStats::deserialize(&mut buf).map_err(Error::from)?; + if !stats.replication_stats.provider_available { + return Err(Error::other("peer replication statistics provider is unavailable")); + } + Ok(stats) +} #[derive(Clone, Debug, PartialEq, Eq)] pub struct ScannerPeerActivity { @@ -819,9 +841,26 @@ impl PeerRestClient { Err(Error::NotImplemented) } - pub async fn get_bucket_stats(&self) -> Result<()> { - warn!("get_bucket_stats is not implemented in PeerRestClient"); - Err(Error::NotImplemented) + pub async fn get_bucket_stats(&self, bucket: &str) -> Result { + let response = self + .finalize_result( + async { + let mut client = self + .get_client() + .await? + .max_decoding_message_size(REPLICATION_STATS_MAX_MESSAGE_SIZE); + let response = client + .get_bucket_stats(Request::new(GetBucketStatsDataRequest { + bucket: bucket.to_string(), + })) + .await? + .into_inner(); + Ok(response) + } + .await, + ) + .await?; + decode_bucket_stats_response(response) } pub async fn get_sr_metrics(&self) -> Result<()> { @@ -1508,6 +1547,50 @@ mod tests { use std::sync::{Arc, Mutex}; use tracing_subscriber::{Registry, fmt::MakeWriter, layer::SubscriberExt}; + #[test] + fn replication_stats_response_decodes_valid_empty_provider() { + let mut stats = BucketStats::default(); + stats.replication_stats.provider_available = true; + let payload = rmp_serde::to_vec_named(&stats).expect("bucket statistics should encode"); + + let decoded = decode_bucket_stats_response(GetBucketStatsDataResponse { + success: true, + bucket_stats: payload.into(), + error_info: None, + }) + .expect("valid bucket statistics should decode"); + + assert!(decoded.replication_stats.provider_available); + assert!(decoded.replication_stats.stats.is_empty()); + } + + #[test] + fn replication_stats_response_rejects_unavailable_malformed_and_oversized_payloads() { + let unavailable = decode_bucket_stats_response(GetBucketStatsDataResponse { + success: false, + bucket_stats: Bytes::new(), + error_info: Some("provider unavailable".to_string()), + }) + .expect_err("unavailable provider must not become a zero snapshot"); + assert!(unavailable.to_string().contains("provider unavailable")); + + let malformed = decode_bucket_stats_response(GetBucketStatsDataResponse { + success: true, + bucket_stats: Bytes::from_static(b"not-msgpack"), + error_info: None, + }) + .expect_err("malformed peer statistics must fail closed"); + assert!(!malformed.to_string().is_empty()); + + let oversized = decode_bucket_stats_response(GetBucketStatsDataResponse { + success: true, + bucket_stats: Bytes::from(vec![0; REPLICATION_STATS_MAX_MESSAGE_SIZE + 1]), + error_info: None, + }) + .expect_err("oversized peer statistics must fail closed"); + assert!(oversized.to_string().contains("size limit")); + } + #[derive(Clone, Default)] struct CapturedLogs { buffer: Arc>>, diff --git a/crates/replication/src/lib.rs b/crates/replication/src/lib.rs index 93a4eb18d..e6759cc49 100644 --- a/crates/replication/src/lib.rs +++ b/crates/replication/src/lib.rs @@ -82,7 +82,7 @@ pub use runtime::{ pub use stats::{ ActiveWorkerStat, BucketReplicationStat, BucketReplicationStats, BucketStats, ExponentialMovingAverage, FailStats, FailedMetric, InQueueMetric, InQueueStats, LatencyStats, ProxyMetric, ProxyStatsCache, QueueCache, QueueNode, QueueStats, - SRMetricsSummary, XferStats, + ReplicationMetricScope, SRMetricsSummary, XferStats, }; pub use storage_api::{DeletedObject, ObjectToDelete}; pub use tagging::{ReplicationTagFilter, decode_tags_to_map}; diff --git a/crates/replication/src/stats.rs b/crates/replication/src/stats.rs index 64452e628..050f164ea 100644 --- a/crates/replication/src/stats.rs +++ b/crates/replication/src/stats.rs @@ -313,18 +313,24 @@ impl InQueueMetric { pub fn merge(&self, other: &InQueueMetric) -> Self { Self { curr: InQueueStats { - bytes: self.curr.bytes + other.curr.bytes, - count: self.curr.count + other.curr.count, + bytes: self.curr.bytes.saturating_add(other.curr.bytes), + count: self.curr.count.saturating_add(other.curr.count), now_bytes: AtomicI64::new( - self.curr.now_bytes.load(Ordering::Relaxed) + other.curr.now_bytes.load(Ordering::Relaxed), + self.curr + .now_bytes + .load(Ordering::Relaxed) + .saturating_add(other.curr.now_bytes.load(Ordering::Relaxed)), ), now_count: AtomicI64::new( - self.curr.now_count.load(Ordering::Relaxed) + other.curr.now_count.load(Ordering::Relaxed), + self.curr + .now_count + .load(Ordering::Relaxed) + .saturating_add(other.curr.now_count.load(Ordering::Relaxed)), ), }, avg: InQueueStats { - bytes: (self.avg.bytes + other.avg.bytes) / 2, - count: (self.avg.count + other.avg.count) / 2, + bytes: self.avg.bytes.saturating_add(other.avg.bytes) / 2, + count: self.avg.count.saturating_add(other.avg.count) / 2, ..Default::default() }, max: InQueueStats { @@ -333,8 +339,8 @@ impl InQueueMetric { ..Default::default() }, last_minute: InQueueStats { - bytes: self.last_minute.bytes + other.last_minute.bytes, - count: self.last_minute.count + other.last_minute.count, + bytes: self.last_minute.bytes.saturating_add(other.last_minute.bytes), + count: self.last_minute.count.saturating_add(other.last_minute.count), ..Default::default() }, samples: VecDeque::new(), @@ -495,8 +501,8 @@ impl FailStats { pub fn add_size(&mut self, size: i64, _err: Option<&E>) { let observed_at = Instant::now(); - self.count += 1; - self.size += size; + self.count = self.count.saturating_add(1); + self.size = self.size.saturating_add(size); self.recent.push_back(FailureSample { observed_at, size }); self.prune(observed_at); } @@ -527,8 +533,8 @@ impl FailStats { pub fn merge(&self, other: &FailStats) -> Self { Self { - count: self.count + other.count, - size: self.size + other.size, + count: self.count.saturating_add(other.count), + size: self.size.saturating_add(other.size), recent: VecDeque::new(), } } @@ -588,6 +594,10 @@ pub struct BucketReplicationStat { pub xfer_rate_sml: XferStats, pub bandwidth_limit_bytes_per_sec: i64, pub current_bandwidth_bytes_per_sec: f64, + #[serde(default)] + pub latency_scope: ReplicationMetricScope, + #[serde(default)] + pub bandwidth_scope: ReplicationMetricScope, } impl BucketReplicationStat { @@ -602,6 +612,22 @@ impl BucketReplicationStat { self.xfer_rate_sml.add_size(size, duration); } } + + pub fn set_node_local_bandwidth(&mut self, limit_bytes_per_sec: i64, current_bytes_per_sec: f64) { + self.bandwidth_limit_bytes_per_sec = limit_bytes_per_sec; + self.current_bandwidth_bytes_per_sec = current_bytes_per_sec; + self.bandwidth_scope = ReplicationMetricScope::NodeLocal; + } +} + +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ReplicationMetricScope { + #[default] + Unavailable, + NodeLocal, + ClusterAggregated, + PartialCluster, } #[derive(Debug, Clone, Default, Serialize, Deserialize)] @@ -632,6 +658,16 @@ pub struct BucketReplicationStats { pub resync_canceled_count: i64, #[serde(default)] pub resync_duration_ms: i64, + #[serde(default)] + pub provider_available: bool, + #[serde(default)] + pub cluster_complete: bool, + #[serde(default)] + pub observed_node_count: u32, + #[serde(default)] + pub expected_node_count: u32, + #[serde(default)] + pub queue_scope: ReplicationMetricScope, } impl BucketReplicationStats { @@ -662,7 +698,18 @@ impl BucketReplicationStats { } pub fn clone_stats(&self) -> Self { - self.clone() + let mut snapshot = self.clone(); + for stat in snapshot.stats.values_mut() { + stat.failed = stat.fail_stats.to_metric(); + } + snapshot + } + + pub fn mark_node_local_provider_available(&mut self) { + self.provider_available = true; + self.cluster_complete = true; + self.observed_node_count = 1; + self.expected_node_count = 1; } pub fn record_resync_status(&mut self, status: ResyncStatusType, duration: Option) { @@ -791,6 +838,27 @@ mod tests { assert_eq!(last_hour.size, 96); } + #[test] + fn fail_stats_saturate_instead_of_wrapping() { + let mut stats = FailStats { + count: i64::MAX, + size: i64::MAX, + ..Default::default() + }; + + stats.add_size(1, None::<&()>); + let merged = stats.merge(&FailStats { + count: 1, + size: 1, + ..Default::default() + }); + + assert_eq!(stats.count, i64::MAX); + assert_eq!(stats.size, i64::MAX); + assert_eq!(merged.count, i64::MAX); + assert_eq!(merged.size, i64::MAX); + } + #[test] fn active_worker_stat_update_tracks_rolling_avg_and_max() { let mut stats = ActiveWorkerStat::default(); diff --git a/rustfs/src/admin/handlers/replication.rs b/rustfs/src/admin/handlers/replication.rs index 74f437170..bf1ff1f07 100644 --- a/rustfs/src/admin/handlers/replication.rs +++ b/rustfs/src/admin/handlers/replication.rs @@ -15,7 +15,10 @@ use crate::admin::auth::validate_admin_request; use crate::admin::handlers::site_replication::site_replication_peer_deployment_id_for_endpoint; use crate::admin::router::{AdminOperation, Operation, S3Router}; -use crate::admin::runtime_sources::{current_replication_stats_handle, current_runtime_port, object_store_from_req}; +use crate::admin::runtime_sources::{ + AppContext, app_context_from_req, current_notification_system_for_context, current_replication_stats_handle_for_context, + current_runtime_port, object_store_from_req, +}; use crate::admin::storage_api::bucket::metadata::BUCKET_TARGETS_FILE; use crate::admin::storage_api::bucket::metadata_sys; use crate::admin::storage_api::bucket::metadata_sys::get_replication_config; @@ -25,6 +28,7 @@ use crate::admin::storage_api::bucket::target_sys::{BucketTargetError, BucketTar use crate::admin::storage_api::contract::bucket::{BucketOperations, BucketOptions}; use crate::admin::storage_api::contract::list::ListOperations as _; use crate::admin::storage_api::error::StorageError; +use crate::admin::storage_api::runtime::PeerRestClient; use crate::admin::utils::read_compatible_admin_body; use crate::auth::{check_key_valid, get_session_token}; use crate::error::ApiError; @@ -39,7 +43,8 @@ use rustfs_policy::policy::action::{Action, AdminAction}; use s3s::header::CONTENT_TYPE; use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error}; use serde::{Deserialize, Serialize}; -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; +use std::sync::Arc; use std::time::Duration; use time::OffsetDateTime; use tracing::{debug, error, info, warn}; @@ -283,6 +288,39 @@ fn is_local_host(_host: String) -> bool { false } +pub(crate) async fn cluster_replication_stats(bucket: &str, context: Option>) -> BucketStats { + let Some(stats) = current_replication_stats_handle_for_context(context.clone()) else { + return BucketStats::default(); + }; + + let local = stats.get_latest_replication_stats(bucket).await; + let Some(notification_system) = current_notification_system_for_context(context.as_deref()) else { + return local; + }; + + let (peers, expected_node_count) = unique_replication_peers(¬ification_system.peer_clients); + let peer_results = futures_util::future::join_all(peers.into_iter().map(|peer| peer.get_bucket_stats(bucket))).await; + let mut snapshots = Vec::with_capacity(peer_results.len().saturating_add(1)); + snapshots.push(local); + snapshots.extend(peer_results.into_iter().filter_map(Result::ok)); + + stats + .aggregate_bucket_replication_stats(bucket, snapshots, expected_node_count) + .await +} + +fn unique_replication_peers(peer_clients: &[Option]) -> (Vec<&PeerRestClient>, u32) { + let mut seen_grid_hosts = HashSet::new(); + let peers: Vec<_> = peer_clients + .iter() + .filter_map(|peer| peer.as_ref()) + .filter(|peer| seen_grid_hosts.insert(peer.grid_host.clone())) + .collect(); + let unavailable_slots = peer_clients.iter().filter(|peer| peer.is_none()).count(); + let expected_node_count = u32::try_from(peers.len().saturating_add(unavailable_slots).saturating_add(1)).unwrap_or(u32::MAX); + (peers, expected_node_count) +} + //awscurl --service s3 --region us-east-1 --access_key rustfsadmin --secret_key rustfsadmin "http://:9000/rustfs/admin/v3/replicationmetrics?bucket=1" pub struct GetReplicationMetricsHandler {} @@ -322,13 +360,7 @@ impl Operation for GetReplicationMetricsHandler { return Err(ApiError::from(err).into()); } - // TODO cluster cache - // In actual implementation, statistics would be obtained from cluster - // This is simplified to get from local cache - let bucket_stats = match current_replication_stats_handle() { - Some(s) => s.get_latest_replication_stats(bucket).await, - None => BucketStats::default(), - }; + let bucket_stats = cluster_replication_stats(bucket, app_context_from_req(&req)).await; let data = serde_json::to_vec(&bucket_stats.replication_stats) .map_err(|_| S3Error::with_message(S3ErrorCode::InternalError, "serialize failed"))?; @@ -779,9 +811,8 @@ impl Operation for ReplicationDiffHandler { } } -/// Failed-replication backlog for one remote target (ARN), summarised from the -/// replication stats. RustFS tracks failed replications as aggregate counters, -/// not an enumerable per-object MRF queue, so this reports the aggregate. +/// Failed-replication totals for one remote target (ARN), summarised from the +/// runtime replication statistics. #[derive(Debug, Serialize)] struct MrfTargetBacklog { #[serde(rename = "ARN")] @@ -790,14 +821,14 @@ struct MrfTargetBacklog { failed_count: i64, #[serde(rename = "FailedSize")] failed_size: i64, + #[serde(rename = "ObservationScope")] + observation_scope: &'static str, } /// Response body for `GET /v3/replication/mrf`. /// -/// MinIO streams individual MRF (most-recently-failed) entries. RustFS persists -/// MRF entries to disk but does not expose a runtime query over that queue, so -/// this endpoint reports the aggregate failed-replication backlog and the -/// in-queue counters instead. See the handler docs and the PR limitations note. +/// Runtime failed/queued totals and the durable MRF recovery backlog are kept +/// separate because persisted MRF entries do not contain a target ARN. #[derive(Debug, Serialize)] struct MrfResponse { #[serde(rename = "Bucket")] @@ -814,18 +845,89 @@ struct MrfResponse { queued_size: i64, #[serde(rename = "PerObjectEntriesAvailable")] per_object_entries_available: bool, + #[serde(rename = "RuntimeStatsAvailable")] + runtime_stats_available: bool, + #[serde(rename = "ClusterComplete")] + cluster_complete: bool, + #[serde(rename = "ObservedNodeCount")] + observed_node_count: u32, + #[serde(rename = "ExpectedNodeCount")] + expected_node_count: u32, + #[serde(rename = "DurableBacklogAvailable")] + durable_backlog_available: bool, + #[serde(rename = "DurableCount")] + durable_count: i64, + #[serde(rename = "DurableSize")] + durable_size: i64, + #[serde(rename = "PerTargetDurableEntriesAvailable")] + per_target_durable_entries_available: bool, +} + +fn build_mrf_response( + bucket: String, + bucket_stats: &BucketStats, + durable: crate::admin::storage_api::replication::DurableMrfBacklog, +) -> MrfResponse { + let observation_scope = if bucket_stats.replication_stats.cluster_complete { + "cluster_aggregated" + } else { + "partial_cluster" + }; + let mut targets: Vec = Vec::with_capacity(bucket_stats.replication_stats.stats.len()); + let mut total_failed_count: i64 = 0; + let mut total_failed_size: i64 = 0; + for (arn, stat) in &bucket_stats.replication_stats.stats { + total_failed_count = total_failed_count.saturating_add(stat.failed.count); + total_failed_size = total_failed_size.saturating_add(stat.failed.size); + targets.push(MrfTargetBacklog { + arn: arn.clone(), + failed_count: stat.failed.count, + failed_size: stat.failed.size, + observation_scope, + }); + } + targets.sort_by(|a, b| a.arn.cmp(&b.arn)); + + let queued = &bucket_stats.replication_stats.q_stat.curr; + let (durable_count, durable_size) = if durable.available { + durable + .entries + .iter() + .filter(|entry| entry.bucket == bucket) + .fold((0i64, 0i64), |(count, size), entry| { + (count.saturating_add(1), size.saturating_add(entry.size)) + }) + } else { + (0, 0) + }; + + MrfResponse { + bucket, + targets, + total_failed_count, + total_failed_size, + queued_count: queued.count, + queued_size: queued.bytes, + per_object_entries_available: false, + runtime_stats_available: bucket_stats.replication_stats.provider_available, + cluster_complete: bucket_stats.replication_stats.cluster_complete, + observed_node_count: bucket_stats.replication_stats.observed_node_count, + expected_node_count: bucket_stats.replication_stats.expected_node_count, + durable_backlog_available: durable.available, + durable_count, + durable_size, + per_target_durable_entries_available: false, + } } /// `GET /v3/replication/mrf` /// /// Reports the failed-replication backlog (MinIO's MRF concept) for a bucket. /// -/// Compatibility note: MinIO returns a stream of individual MRF entries drained -/// from its in-memory MRF queue. RustFS records failed replications as aggregate -/// counters in the replication stats and persists MRF entries to disk without a -/// runtime query API, so this handler returns the aggregate failed and queued -/// counts per target instead of per-object rows. `PerObjectEntriesAvailable` is -/// always `false` to make that limitation explicit to clients. +/// Compatibility note: MinIO returns a stream of individual MRF entries. RustFS +/// deliberately returns aggregate runtime and durable counters instead. +/// `PerObjectEntriesAvailable` and `PerTargetDurableEntriesAvailable` remain +/// false until an enumerable API and target-bearing durable format exist. pub struct ReplicationMrfHandler {} #[async_trait::async_trait] @@ -857,35 +959,9 @@ impl Operation for ReplicationMrfHandler { return Err(ApiError::from(err).into()); } - let bucket_stats = match current_replication_stats_handle() { - Some(s) => s.get_latest_replication_stats(&bucket).await, - None => BucketStats::default(), - }; - - let mut targets: Vec = Vec::new(); - let mut total_failed_count: i64 = 0; - let mut total_failed_size: i64 = 0; - for (arn, stat) in &bucket_stats.replication_stats.stats { - total_failed_count += stat.failed.count; - total_failed_size += stat.failed.size; - targets.push(MrfTargetBacklog { - arn: arn.clone(), - failed_count: stat.failed.count, - failed_size: stat.failed.size, - }); - } - targets.sort_by(|a, b| a.arn.cmp(&b.arn)); - - let queued = &bucket_stats.replication_stats.q_stat.curr; - let response = MrfResponse { - bucket, - targets, - total_failed_count, - total_failed_size, - queued_count: queued.count, - queued_size: queued.bytes, - per_object_entries_available: false, - }; + let durable = crate::admin::storage_api::replication::read_durable_mrf_backlog(store).await; + let bucket_stats = cluster_replication_stats(&bucket, app_context_from_req(&req)).await; + let response = build_mrf_response(bucket, &bucket_stats, durable); let data = serde_json::to_vec(&response) .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("serialize failed: {e}")))?; @@ -897,8 +973,12 @@ impl Operation for ReplicationMrfHandler { #[cfg(test)] mod tests { - use super::{RemoteTargetRequest, extract_query_params, validate_remote_target_tls_settings}; + use super::{ + RemoteTargetRequest, build_mrf_response, extract_query_params, unique_replication_peers, + validate_remote_target_tls_settings, + }; use crate::admin::storage_api::bucket::target::BucketTarget; + use crate::admin::storage_api::replication::{BucketStats, DurableMrfBacklog, MrfOpKind, MrfReplicateEntry}; use http::Uri; fn valid_remote_target_request() -> serde_json::Value { @@ -914,6 +994,115 @@ mod tests { }) } + #[test] + fn cluster_peer_plan_deduplicates_nodes_and_counts_unavailable_slots() { + let peer = crate::admin::storage_api::runtime::PeerRestClient::new( + rustfs_utils::XHost::try_from("127.0.0.1:9000".to_string()).expect("peer host should parse"), + "node-a.example.com:9001".to_string(), + ); + let slots = vec![Some(peer.clone()), Some(peer), None]; + + let (peers, expected_node_count) = unique_replication_peers(&slots); + + assert_eq!(peers.len(), 1); + assert_eq!(peers[0].grid_host, "node-a.example.com:9001"); + assert_eq!(expected_node_count, 3); + } + + #[test] + fn mrf_response_keeps_runtime_and_durable_truth_separate() { + let mut stats = BucketStats::default(); + stats.replication_stats.provider_available = true; + stats.replication_stats.cluster_complete = false; + stats.replication_stats.observed_node_count = 2; + stats.replication_stats.expected_node_count = 3; + let target = stats.replication_stats.stats.entry("arn-a".to_string()).or_default(); + target.failed.count = 3; + target.failed.size = 900; + stats + .replication_stats + .q_stat + .curr + .now_count + .store(4, std::sync::atomic::Ordering::Relaxed); + stats + .replication_stats + .q_stat + .curr + .now_bytes + .store(1200, std::sync::atomic::Ordering::Relaxed); + stats.replication_stats.q_stat = stats.replication_stats.q_stat.snapshot(); + let durable = DurableMrfBacklog { + available: true, + entries: vec![ + MrfReplicateEntry { + bucket: "bucket-a".to_string(), + object: "object-a".to_string(), + version_id: None, + retry_count: 0, + size: 250, + op: MrfOpKind::Object, + delete_marker_version_id: None, + delete_marker: false, + delete_marker_mtime: None, + }, + MrfReplicateEntry { + bucket: "other-bucket".to_string(), + object: "object-b".to_string(), + version_id: None, + retry_count: 0, + size: 999, + op: MrfOpKind::Object, + delete_marker_version_id: None, + delete_marker: false, + delete_marker_mtime: None, + }, + ], + }; + + let response = build_mrf_response("bucket-a".to_string(), &stats, durable); + let json = serde_json::to_value(response).expect("MRF response should serialize"); + + assert_eq!(json["TotalFailedCount"], 3); + assert_eq!(json["TotalFailedSize"], 900); + assert_eq!(json["QueuedCount"], 4); + assert_eq!(json["QueuedSize"], 1200); + assert_eq!(json["DurableCount"], 1); + assert_eq!(json["DurableSize"], 250); + assert_eq!(json["RuntimeStatsAvailable"], true); + assert_eq!(json["ClusterComplete"], false); + assert_eq!(json["Targets"][0]["ObservationScope"], "partial_cluster"); + assert_eq!(json["PerObjectEntriesAvailable"], false); + assert_eq!(json["PerTargetDurableEntriesAvailable"], false); + } + + #[test] + fn mrf_response_distinguishes_unavailable_sources_from_valid_zero() { + let unavailable = build_mrf_response("bucket-a".to_string(), &BucketStats::default(), DurableMrfBacklog::default()); + let unavailable_json = serde_json::to_value(unavailable).expect("unavailable response should serialize"); + assert_eq!(unavailable_json["RuntimeStatsAvailable"], false); + assert_eq!(unavailable_json["DurableBacklogAvailable"], false); + + let mut valid_empty_stats = BucketStats::default(); + valid_empty_stats.replication_stats.provider_available = true; + valid_empty_stats.replication_stats.cluster_complete = true; + valid_empty_stats.replication_stats.observed_node_count = 1; + valid_empty_stats.replication_stats.expected_node_count = 1; + let valid_empty = build_mrf_response( + "bucket-a".to_string(), + &valid_empty_stats, + DurableMrfBacklog { + available: true, + entries: Vec::new(), + }, + ); + let valid_empty_json = serde_json::to_value(valid_empty).expect("valid empty response should serialize"); + assert_eq!(valid_empty_json["RuntimeStatsAvailable"], true); + assert_eq!(valid_empty_json["DurableBacklogAvailable"], true); + assert_eq!(valid_empty_json["TotalFailedCount"], 0); + assert_eq!(valid_empty_json["DurableCount"], 0); + } + #[test] fn test_extract_query_params_decodes_percent_encoded_values() { let uri: Uri = "/rustfs/admin/v3/list-remote-targets?bucket=foo%2Fbar&flag=a+b" diff --git a/rustfs/src/admin/router.rs b/rustfs/src/admin/router.rs index 15aa68454..e60fbe4aa 100644 --- a/rustfs/src/admin/router.rs +++ b/rustfs/src/admin/router.rs @@ -28,8 +28,8 @@ use super::storage_api::runtime::PeerRestClient; use crate::admin::console::{is_console_path, make_console_server}; use crate::admin::handlers::oidc::is_oidc_path; use crate::admin::runtime_sources::{ - ServerContextSlot, current_boot_time, current_bucket_monitor_handle, current_deployment_id, current_notification_system, - current_object_store_handle, current_region, current_replication_pool_handle, current_replication_stats_handle, + ServerContextSlot, app_context_from_req, current_boot_time, current_bucket_monitor_handle, current_deployment_id, + current_notification_system, current_object_store_handle, current_region, current_replication_pool_handle, current_server_config, default_object_usecase, }; use crate::admin::storage_api::access::{ReqInfo, authorize_request, spawn_traced}; @@ -1509,11 +1509,12 @@ async fn ensure_replication_config_exists(bucket: &str) -> S3Result<()> { } } -async fn build_replication_metrics_response(bucket: &str, route: ReplicationExtRoute) -> S3Result> { - let bucket_stats = match current_replication_stats_handle() { - Some(stats) => stats.get_latest_replication_stats(bucket).await, - None => BucketStats::default(), - }; +async fn build_replication_metrics_response( + bucket: &str, + route: ReplicationExtRoute, + context: Option>, +) -> S3Result> { + let bucket_stats = crate::admin::handlers::replication::cluster_replication_stats(bucket, context).await; let bucket_stats = apply_replication_metrics_bandwidth_report(bucket_stats, collect_replication_metrics_bandwidth(bucket)); let bucket_stats = apply_replication_metrics_runtime_fields(bucket_stats, route, replication_metrics_uptime_seconds()); @@ -1555,10 +1556,12 @@ fn apply_replication_metrics_bandwidth_report( mut bucket_stats: BucketStats, bandwidth_report: HashMap, ) -> BucketStats { + if bucket_stats.replication_stats.expected_node_count > 1 { + return bucket_stats; + } for (arn, details) in bandwidth_report { let stat = bucket_stats.replication_stats.stats.entry(arn).or_default(); - stat.bandwidth_limit_bytes_per_sec = details.limit_bytes_per_sec; - stat.current_bandwidth_bytes_per_sec = details.current_bandwidth_bytes_per_sec; + stat.set_node_local_bandwidth(details.limit_bytes_per_sec, details.current_bandwidth_bytes_per_sec); } bucket_stats @@ -2519,7 +2522,7 @@ async fn handle_replication_extension_request( match ext_req.route { ReplicationExtRoute::MetricsV1 | ReplicationExtRoute::MetricsV2 => { ensure_replication_config_exists(&ext_req.bucket).await?; - build_replication_metrics_response(&ext_req.bucket, ext_req.route).await + build_replication_metrics_response(&ext_req.bucket, ext_req.route, app_context_from_req(req)).await } ReplicationExtRoute::Check => { let (versioning, _) = metadata_sys::get_versioning_config(&ext_req.bucket) @@ -3738,6 +3741,10 @@ mod tests { assert_eq!(stat.replicated_count, 3); assert_eq!(stat.bandwidth_limit_bytes_per_sec, 2048); assert_eq!(stat.current_bandwidth_bytes_per_sec, 1536.5); + assert_eq!( + serde_json::to_value(stat).expect("target stats should serialize")["bandwidth_scope"], + "node_local" + ); } #[test] @@ -3761,6 +3768,34 @@ mod tests { assert_eq!(stat.current_bandwidth_bytes_per_sec, 1024.25); } + #[test] + fn apply_replication_metrics_bandwidth_report_does_not_replace_cluster_values() { + let mut stats = BucketStats::default(); + stats.replication_stats.expected_node_count = 2; + let target = stats + .replication_stats + .stats + .entry("arn:replication:a".to_string()) + .or_default(); + target.bandwidth_limit_bytes_per_sec = 8192; + target.current_bandwidth_bytes_per_sec = 3000.0; + + let updated = apply_replication_metrics_bandwidth_report( + stats, + HashMap::from([( + "arn:replication:a".to_string(), + BandwidthDetails { + limit_bytes_per_sec: 2048, + current_bandwidth_bytes_per_sec: 1000.0, + }, + )]), + ); + let target = &updated.replication_stats.stats["arn:replication:a"]; + + assert_eq!(target.bandwidth_limit_bytes_per_sec, 8192); + assert_eq!(target.current_bandwidth_bytes_per_sec, 3000.0); + } + #[test] fn serialize_replication_metrics_body_v2_returns_full_bucket_stats() { let mut stats = BucketStats { diff --git a/rustfs/src/admin/runtime_sources.rs b/rustfs/src/admin/runtime_sources.rs index 919ede680..7f02f8c65 100644 --- a/rustfs/src/admin/runtime_sources.rs +++ b/rustfs/src/admin/runtime_sources.rs @@ -26,8 +26,8 @@ pub(crate) use crate::runtime_sources::{ current_bucket_monitor_handle, current_deployment_id, current_endpoints_handle, current_federated_identity_service, current_iam_handle, current_kms_runtime_service_manager, current_notification_system_for_context, current_object_data_cache_handle_for_context, current_object_store_handle_for_context, current_ready_iam_handle, - current_region, current_replication_pool_handle, current_replication_stats_handle, current_server_config_for_context, - current_token_signing_key, + current_region, current_replication_pool_handle, current_replication_stats_handle, + current_replication_stats_handle_for_context, current_server_config_for_context, current_token_signing_key, }; #[cfg(test)] pub(crate) use crate::runtime_sources::{IamInterface, KmsInterface, ServerConfigInterface, StorageClassInterface}; diff --git a/rustfs/src/admin/storage_api.rs b/rustfs/src/admin/storage_api.rs index 33c6f9651..f875094e7 100644 --- a/rustfs/src/admin/storage_api.rs +++ b/rustfs/src/admin/storage_api.rs @@ -340,6 +340,15 @@ pub(crate) mod replication { pub(crate) type ResyncOpts = super::ecstore_bucket::replication::ResyncOpts; pub(crate) type ResyncStatusType = super::ecstore_bucket::replication::ResyncStatusType; pub(crate) type TargetReplicationResyncStatus = super::ecstore_bucket::replication::TargetReplicationResyncStatus; + pub(crate) type DurableMrfBacklog = super::ecstore_bucket::replication::DurableMrfBacklog; + #[cfg(test)] + pub(crate) type MrfReplicateEntry = super::ecstore_bucket::replication::MrfReplicateEntry; + #[cfg(test)] + pub(crate) type MrfOpKind = super::ecstore_bucket::replication::MrfOpKind; + + pub(crate) async fn read_durable_mrf_backlog(api: std::sync::Arc) -> DurableMrfBacklog { + super::ecstore_bucket::replication::read_durable_mrf_backlog(api).await + } pub(crate) fn resync_opts( bucket: &str, diff --git a/rustfs/src/app/context.rs b/rustfs/src/app/context.rs index 65cbf2fbb..9e6da5770 100644 --- a/rustfs/src/app/context.rs +++ b/rustfs/src/app/context.rs @@ -366,7 +366,7 @@ fn resolve_replication_pool_handle_with(context: Option>) -> Opt context.and_then(|context| context.replication_pool().handle()) } -fn resolve_replication_stats_handle_with(context: Option>) -> Option> { +pub(crate) fn resolve_replication_stats_handle_with(context: Option>) -> Option> { context.and_then(|context| context.replication_stats().handle()) } diff --git a/rustfs/src/runtime_sources.rs b/rustfs/src/runtime_sources.rs index b1fc8b09b..c8575828a 100644 --- a/rustfs/src/runtime_sources.rs +++ b/rustfs/src/runtime_sources.rs @@ -44,9 +44,11 @@ pub(crate) use context::{ resolve_outbound_tls_generation as current_outbound_tls_generation, resolve_outbound_tls_state as current_outbound_tls_state, resolve_performance_metrics as current_performance_metrics, resolve_ready_iam_handle as current_ready_iam_handle, resolve_region as current_region, resolve_replication_pool_handle as current_replication_pool_handle, - resolve_replication_stats_handle as current_replication_stats_handle, resolve_runtime_port as current_runtime_port, - resolve_s3select_db as current_s3select_db, resolve_scanner_metrics_report as current_scanner_metrics_report, - resolve_server_config as current_server_config, resolve_server_config_for_context as current_server_config_for_context, + resolve_replication_stats_handle as current_replication_stats_handle, + resolve_replication_stats_handle_with as current_replication_stats_handle_for_context, + resolve_runtime_port as current_runtime_port, resolve_s3select_db as current_s3select_db, + resolve_scanner_metrics_report as current_scanner_metrics_report, resolve_server_config as current_server_config, + resolve_server_config_for_context as current_server_config_for_context, resolve_tier_config_handle as current_tier_config_handle, resolve_token_signing_key as current_token_signing_key, }; diff --git a/rustfs/src/storage/rpc/node_service.rs b/rustfs/src/storage/rpc/node_service.rs index b82abb52b..2f8d6d411 100644 --- a/rustfs/src/storage/rpc/node_service.rs +++ b/rustfs/src/storage/rpc/node_service.rs @@ -1209,9 +1209,28 @@ impl Node for NodeService { async fn get_bucket_stats( &self, - _request: Request, + request: Request, ) -> Result, Status> { - Err(unimplemented_rpc("get_bucket_stats")) + let bucket = request.into_inner().bucket; + if bucket.is_empty() { + return Err(Status::invalid_argument("bucket is required")); + } + let context = self.context.clone().or_else(runtime_sources::current_app_context); + let Some(stats) = runtime_sources::current_replication_stats_handle_for_context(context.as_deref()) else { + return Ok(Response::new(GetBucketStatsDataResponse { + success: false, + bucket_stats: Bytes::new(), + error_info: Some("replication statistics provider is unavailable".to_string()), + })); + }; + let bucket_stats = stats.get_latest_replication_stats(&bucket).await; + let bucket_stats = + rmp_serde::to_vec_named(&bucket_stats).map_err(|_| Status::internal("failed to serialize replication statistics"))?; + Ok(Response::new(GetBucketStatsDataResponse { + success: true, + bucket_stats: bucket_stats.into(), + error_info: None, + })) } async fn get_sr_metrics( @@ -4428,12 +4447,11 @@ mod tests { .await, "download_profile_data", ); - assert_unimplemented_status( - service - .get_bucket_stats(Request::new(GetBucketStatsDataRequest::default())) - .await, - "get_bucket_stats", - ); + let bucket_stats_err = service + .get_bucket_stats(Request::new(GetBucketStatsDataRequest::default())) + .await + .expect_err("empty bucket statistics request should fail"); + assert_eq!(bucket_stats_err.code(), tonic::Code::InvalidArgument); assert_unimplemented_status( service.get_sr_metrics(Request::new(GetSrMetricsDataRequest::default())).await, "get_sr_metrics", diff --git a/rustfs/src/storage/runtime_sources.rs b/rustfs/src/storage/runtime_sources.rs index 49aca244d..6c1d556b5 100644 --- a/rustfs/src/storage/runtime_sources.rs +++ b/rustfs/src/storage/runtime_sources.rs @@ -39,6 +39,12 @@ pub(crate) fn current_object_store_handle_for_context(context: Option<&AppContex root_runtime_sources::current_object_store_handle_for_context(context) } +pub(crate) fn current_replication_stats_handle_for_context( + context: Option<&AppContext>, +) -> Option> { + root_runtime_sources::current_replication_stats_handle_for_context(context.map(|context| Arc::new(context.clone()))) +} + pub(crate) fn current_buffer_config() -> RustFSBufferConfig { root_runtime_sources::current_buffer_config().unwrap_or_default() } diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index 0c9381a50..413412628 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -699,6 +699,17 @@ impl StorageReplicationStatsHandle { self.inner.get_latest_replication_stats(bucket).await } + pub(crate) async fn aggregate_bucket_replication_stats( + &self, + bucket: &str, + stats: Vec, + expected_node_count: u32, + ) -> ecstore_bucket::replication::BucketStats { + self.inner + .aggregate_bucket_replication_stats(bucket, stats, expected_node_count) + .await + } + pub(crate) async fn site_metrics_snapshot(&self) -> ReplicationSiteMetricsSnapshot { let metrics = self.inner.get_sr_metrics_for_node().await; ReplicationSiteMetricsSnapshot {