diff --git a/crates/e2e_test/src/replication_extension_test.rs b/crates/e2e_test/src/replication_extension_test.rs index 8c6ff2a04..07b0e7040 100644 --- a/crates/e2e_test/src/replication_extension_test.rs +++ b/crates/e2e_test/src/replication_extension_test.rs @@ -29,8 +29,9 @@ use aws_sdk_s3::types::{ use aws_sdk_s3::{Client, Config}; use base64::{Engine, engine::general_purpose::STANDARD as BASE64_STANDARD}; use bytes::Bytes; +use flate2::read::GzDecoder; use futures::{Stream, StreamExt}; -use http::header::{CONTENT_TYPE, HOST}; +use http::header::{CONTENT_ENCODING, CONTENT_TYPE, HOST}; use http_body_util::{BodyExt, Full}; use hyper::body::Incoming; use hyper::server::conn::http1; @@ -38,6 +39,10 @@ use hyper::service::service_fn; use hyper::{Request, Response}; use hyper_util::rt::TokioIo; use local_ip_address::local_ip; +use opentelemetry_proto::tonic::collector::metrics::v1::ExportMetricsServiceRequest; +use opentelemetry_proto::tonic::common::v1::{KeyValue, any_value::Value as AnyValue}; +use opentelemetry_proto::tonic::metrics::v1::{Metric, metric, number_data_point}; +use prost::Message; use rcgen::{ BasicConstraints, CertificateParams, CertifiedIssuer, DnType, ExtendedKeyUsagePurpose, IsCa, KeyPair, KeyUsagePurpose, SanType, generate_simple_self_signed, @@ -56,6 +61,7 @@ use sha2::{Digest, Sha256}; use std::collections::BTreeMap; use std::convert::Infallible; use std::error::Error; +use std::io::Read; use std::net::IpAddr; use std::path::Path; use std::process::Command; @@ -64,11 +70,13 @@ use std::sync::atomic::{AtomicU64, Ordering}; use time::{Duration as TimeDuration, OffsetDateTime}; use tokio::fs; use tokio::net::TcpListener; -use tokio::sync::watch; +use tokio::sync::{Mutex, watch}; +use tokio::task::JoinHandle; use tokio::task::JoinSet; use tokio::time::{Duration, sleep, timeout}; type TestResult = Result<(), Box>; +type BacklogMetricPoints = Arc>>>; /// A replication source server validates the remote target endpoint, and the e2e /// target runs on loopback (127.0.0.1), which RustFS's SSRF egress guard rejects by @@ -107,6 +115,250 @@ const REPL17_KMS_KEY_ID: &str = "repl17-local-key"; const REPL17_SSEC_KEY: &str = "01234567890123456789012345678901"; const REPLICATION_FAILED_EVENT: &str = "s3:Replication:OperationFailedReplication"; const REPLICATION_EVENT_MAX_BUFFER_BYTES: usize = 1024 * 1024; +const OTLP_METRICS_BODY_LIMIT: u64 = 4 * 1024 * 1024; +const BUCKET_LABEL: &str = "bucket"; +const CURRENT_BACKLOG_COUNT_METRIC: &str = "rustfs_bucket_replication_current_backlog_count"; +const CURRENT_BACKLOG_BYTES_METRIC: &str = "rustfs_bucket_replication_current_backlog_bytes"; +const MRF_PENDING_COUNT_METRIC: &str = "rustfs_bucket_replication_mrf_pending_count"; +const MRF_PENDING_BYTES_METRIC: &str = "rustfs_bucket_replication_mrf_pending_bytes"; + +struct ReplicationBacklogMetricCollector { + endpoint: String, + values: BacklogMetricPoints, + task: JoinHandle<()>, +} + +impl ReplicationBacklogMetricCollector { + async fn start() -> Result> { + let listener = TcpListener::bind("127.0.0.1:0").await?; + let endpoint = format!("http://{}/v1/metrics", listener.local_addr()?); + let values = Arc::new(Mutex::new(BTreeMap::new())); + let task_values = values.clone(); + let task = tokio::spawn(async move { + loop { + let Ok((stream, _)) = listener.accept().await else { + break; + }; + let values = task_values.clone(); + tokio::spawn(async move { + let _ = http1::Builder::new() + .serve_connection( + TokioIo::new(stream), + service_fn(move |request| handle_backlog_metric_export(request, values.clone())), + ) + .await; + }); + } + }); + + Ok(Self { endpoint, values, task }) + } + + fn root_endpoint(&self) -> &str { + self.endpoint.trim_end_matches("/v1/metrics") + } + + async fn bucket_metric_value(&self, metric: &str, bucket: &str) -> f64 { + self.values + .lock() + .await + .get(metric) + .and_then(|buckets| buckets.get(bucket)) + .map(|(_, value)| *value) + .unwrap_or_default() + } + + async fn wait_for_bucket_metric( + &self, + metric: &str, + bucket: &str, + expected: impl Fn(f64) -> bool, + description: &str, + ) -> Result> { + let deadline = tokio::time::Instant::now() + Duration::from_secs(30); + loop { + let value = self.bucket_metric_value(metric, bucket).await; + if expected(value) { + return Ok(value); + } + if tokio::time::Instant::now() >= deadline { + let snapshot = self.values.lock().await.clone(); + return Err(format!("timed out waiting for {metric} on bucket {bucket} to satisfy {description}; last={value}, snapshot={snapshot:?}").into()); + } + sleep(Duration::from_millis(200)).await; + } + } +} + +impl Drop for ReplicationBacklogMetricCollector { + fn drop(&mut self) { + self.task.abort(); + } +} + +async fn handle_backlog_metric_export( + request: Request, + values: BacklogMetricPoints, +) -> Result>, Infallible> { + if request.uri().path() != "/v1/metrics" { + return Ok(empty_http_response(StatusCode::NOT_FOUND)); + } + + let gzip = request + .headers() + .get(CONTENT_ENCODING) + .and_then(|value| value.to_str().ok()) + .is_some_and(|value| value.eq_ignore_ascii_case("gzip")); + let Ok(collected) = request.into_body().collect().await else { + return Ok(empty_http_response(StatusCode::BAD_REQUEST)); + }; + let body = collected.to_bytes(); + if body.len() as u64 > OTLP_METRICS_BODY_LIMIT { + return Ok(empty_http_response(StatusCode::PAYLOAD_TOO_LARGE)); + } + let payload = if gzip { + let mut decoder = GzDecoder::new(body.as_ref()); + let mut decoded = Vec::new(); + if decoder + .by_ref() + .take(OTLP_METRICS_BODY_LIMIT + 1) + .read_to_end(&mut decoded) + .is_err() + || decoded.len() as u64 > OTLP_METRICS_BODY_LIMIT + { + return Ok(empty_http_response(StatusCode::BAD_REQUEST)); + } + decoded + } else { + body.to_vec() + }; + + match ExportMetricsServiceRequest::decode(payload.as_slice()) { + Ok(export) => { + let mut values = values.lock().await; + record_backlog_metrics(&export, &mut values); + Ok(empty_http_response(StatusCode::OK)) + } + Err(_) => Ok(empty_http_response(StatusCode::BAD_REQUEST)), + } +} + +fn empty_http_response(status: StatusCode) -> Response> { + Response::builder() + .status(status) + .body(Full::new(Bytes::new())) + .expect("static HTTP response is valid") +} + +fn record_backlog_metrics(export: &ExportMetricsServiceRequest, values: &mut BTreeMap>) { + for resource_metrics in &export.resource_metrics { + for scope_metrics in &resource_metrics.scope_metrics { + for metric in &scope_metrics.metrics { + record_backlog_metric(metric, values); + } + } + } +} + +fn record_backlog_metric(metric: &Metric, values: &mut BTreeMap>) { + if ![ + CURRENT_BACKLOG_COUNT_METRIC, + CURRENT_BACKLOG_BYTES_METRIC, + MRF_PENDING_COUNT_METRIC, + MRF_PENDING_BYTES_METRIC, + ] + .contains(&metric.name.as_str()) + { + return; + } + + let points = match &metric.data { + Some(metric::Data::Gauge(gauge)) => gauge.data_points.as_slice(), + Some(metric::Data::Sum(sum)) => sum.data_points.as_slice(), + _ => return, + }; + for point in points { + let Some(bucket) = attribute_string(&point.attributes, BUCKET_LABEL) else { + continue; + }; + let Some(value) = number_point_value(point.value.as_ref()) else { + continue; + }; + values + .entry(metric.name.clone()) + .or_default() + .entry(bucket.to_string()) + .and_modify(|current| { + if point.time_unix_nano >= current.0 { + *current = (point.time_unix_nano, value); + } + }) + .or_insert((point.time_unix_nano, value)); + } +} + +fn number_point_value(value: Option<&number_data_point::Value>) -> Option { + match value? { + number_data_point::Value::AsDouble(value) => Some(*value), + number_data_point::Value::AsInt(value) => Some(*value as f64), + } +} + +fn attribute_string<'a>(attributes: &'a [KeyValue], wanted_key: &str) -> Option<&'a str> { + attributes.iter().find_map(|attribute| { + if attribute.key != wanted_key { + return None; + } + match attribute.value.as_ref()?.value.as_ref()? { + AnyValue::StringValue(value) => Some(value.as_str()), + _ => None, + } + }) +} + +struct SlowReplicationTargetGuard { + task: Option>, +} + +impl SlowReplicationTargetGuard { + async fn bind(address: &str, response_delay: Duration) -> Result> { + let listener = TcpListener::bind(address).await?; + let task = tokio::spawn(async move { + loop { + let Ok((stream, _)) = listener.accept().await else { + break; + }; + tokio::spawn(async move { + let _ = http1::Builder::new() + .serve_connection( + TokioIo::new(stream), + service_fn(move |_request| async move { + sleep(response_delay).await; + Ok::<_, Infallible>(empty_http_response(StatusCode::SERVICE_UNAVAILABLE)) + }), + ) + .await; + }); + } + }); + Ok(Self { task: Some(task) }) + } + + async fn stop(mut self) { + if let Some(task) = self.task.take() { + task.abort(); + let _ = task.await; + } + } +} + +impl Drop for SlowReplicationTargetGuard { + fn drop(&mut self) { + if let Some(task) = self.task.take() { + task.abort(); + } + } +} #[derive(Debug, Clone, serde::Deserialize)] struct ReplicationResetStatusResponse { @@ -3659,6 +3911,125 @@ async fn test_bucket_replication_recovers_after_target_outage() -> TestResult { Ok(()) } +/// backlog#1610 - black-box bucket replication backlog observability. +/// +/// The source exports metrics through the same OTLP path production uses. A slow +/// loopback target keeps replication workers occupied long enough for the metrics +/// runtime to publish non-zero bucket backlog gauges; after the real target +/// returns, replication must converge and the exported current/MRF pending gauges +/// must settle back to zero. +#[tokio::test] +#[serial] +async fn test_bucket_replication_backlog_metrics_observe_outage_and_recovery() -> TestResult { + init_logging(); + + let collector = ReplicationBacklogMetricCollector::start().await?; + let mut source_env = RustFSTestEnvironment::new().await?; + let metric_root = collector.root_endpoint().to_string(); + let metric_endpoint = collector.endpoint.clone(); + let mut source_env_vars: Vec<(&str, &str)> = replication_fast_env().into_iter().collect(); + source_env_vars.extend_from_slice(LOOPBACK_REPLICATION_TARGET_ENV); + source_env_vars.extend_from_slice(FAST_SCANNER_ENV); + source_env_vars.extend_from_slice(&[ + ("RUSTFS_OBS_ENDPOINT", metric_root.as_str()), + ("RUSTFS_OBS_METRIC_ENDPOINT", metric_endpoint.as_str()), + ("RUSTFS_OBS_METRICS_EXPORT_ENABLED", "true"), + ("RUSTFS_OBS_TRACES_EXPORT_ENABLED", "false"), + ("RUSTFS_OBS_LOGS_EXPORT_ENABLED", "false"), + ("RUSTFS_OBS_METER_INTERVAL", "1"), + ("RUSTFS_OBS_USE_STDOUT", "false"), + ("RUSTFS_METRICS_BUCKET_REPLICATION_BANDWIDTH_INTERVAL_SEC", "1"), + ]); + source_env.start_rustfs_server_with_env(vec![], &source_env_vars).await?; + + let mut target_env = RustFSTestEnvironment::new().await?; + target_env.start_rustfs_server_without_cleanup(vec![]).await?; + + let source_bucket = "repl-backlog-metrics-src"; + let target_bucket = "repl-backlog-metrics-dst"; + let source_client = source_env.create_s3_client(); + let target_client = target_env.create_s3_client(); + + source_client.create_bucket().bucket(source_bucket).send().await?; + target_client.create_bucket().bucket(target_bucket).send().await?; + enable_bucket_versioning(&source_env, source_bucket).await?; + enable_bucket_versioning(&target_env, target_bucket).await?; + + let target_arn = set_replication_target(&source_env, source_bucket, &target_env, target_bucket).await?; + put_bucket_replication(&source_env, source_bucket, &target_arn).await?; + + source_client + .put_object() + .bucket(source_bucket) + .key("before-outage.txt") + .body(ByteStream::from_static(b"baseline written before backlog metrics outage")) + .send() + .await?; + assert_replication_converged(&source_client, source_bucket, &target_client, target_bucket).await?; + + target_env.stop_server(); + let slow_target = SlowReplicationTargetGuard::bind(&target_env.address, Duration::from_secs(5)).await?; + + let outage_keys = ["metrics-outage-1.txt", "metrics-outage-2.txt", "metrics-outage-3.txt"]; + for key in outage_keys { + source_client + .put_object() + .bucket(source_bucket) + .key(key) + .body(ByteStream::from( + format!("written while backlog metrics target was slow: {key}").into_bytes(), + )) + .send() + .await?; + } + + let current_backlog = collector + .wait_for_bucket_metric( + CURRENT_BACKLOG_COUNT_METRIC, + source_bucket, + |value| value >= 1.0, + "be at least 1 during outage", + ) + .await?; + let current_bytes = collector + .wait_for_bucket_metric( + CURRENT_BACKLOG_BYTES_METRIC, + source_bucket, + |value| value > 0.0, + "report bytes during outage", + ) + .await?; + assert!( + current_bytes >= current_backlog, + "backlog bytes should be at least the object count while queued; count={current_backlog}, bytes={current_bytes}" + ); + + slow_target.stop().await; + target_env.restart_server_preserving_data(vec![], &[]).await?; + + assert_replication_converged(&source_client, source_bucket, &target_client, target_bucket).await?; + for metric in [ + CURRENT_BACKLOG_COUNT_METRIC, + CURRENT_BACKLOG_BYTES_METRIC, + MRF_PENDING_COUNT_METRIC, + MRF_PENDING_BYTES_METRIC, + ] { + collector + .wait_for_bucket_metric(metric, source_bucket, |value| value == 0.0, "settle back to zero after recovery") + .await?; + } + + let target_state = list_replication_state(&target_client, target_bucket).await?; + for key in ["before-outage.txt"].into_iter().chain(outage_keys) { + assert!( + target_state.iter().any(|entry| entry.key == key && !entry.delete_marker), + "target missing object {key} after backlog metrics recovery; state={target_state:?}" + ); + } + + Ok(()) +} + /// backlog#1147 repl-5, scenario (b) — failure state survives a source restart /// (mirrors backlog#858 delete-decision re-derivation and #859 no-drop). /// diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index b2af30c62..2f86d3ca9 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -173,7 +173,8 @@ pub mod bucket { pub mod replication { pub use crate::bucket::replication::replication_pool::{ - DurableMrfBacklogSummary, DurableMrfBucketBacklog, durable_mrf_backlog_summary_snapshot, + DurableMrfBacklogSummary, DurableMrfBucketBacklog, MrfBacklogObservabilitySummary, MrfBucketBacklogObservability, + durable_mrf_backlog_summary_snapshot, mrf_backlog_observability_snapshot, }; pub use crate::bucket::replication::{ BucketReplicationResyncStatus, BucketStats, DeletedObjectReplicationInfo, DurableMrfBacklog, DynReplicationPool, diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs index 245590977..67afcc5f1 100644 --- a/crates/ecstore/src/bucket/replication/replication_pool.rs +++ b/crates/ecstore/src/bucket/replication/replication_pool.rs @@ -53,6 +53,7 @@ use std::sync::LazyLock; use std::sync::RwLock as StdRwLock; use std::sync::atomic::AtomicI32; use std::sync::atomic::Ordering; +use std::time::Instant; use time::OffsetDateTime; use time::format_description::well_known::Rfc3339; use tokio::sync::Mutex; @@ -91,20 +92,46 @@ pub struct DurableMrfBacklogSummary { pub buckets: Vec, } +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct MrfBucketBacklogObservability { + pub bucket: String, + pub pending_count: u64, + pub pending_bytes: u64, + pub dropped_count: u64, + pub missed_count: u64, + pub flush_failure_count: u64, + pub last_flush_duration_millis: u64, +} + +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct MrfBacklogObservabilitySummary { + pub buckets: Vec, +} + static DURABLE_MRF_BACKLOG_SUMMARY: LazyLock> = LazyLock::new(|| StdRwLock::new(DurableMrfBacklogSummary::default())); +static MRF_BACKLOG_OBSERVABILITY: LazyLock> = + LazyLock::new(|| StdRwLock::new(MrfBacklogObservabilityTracker::default())); -fn durable_mrf_backlog_summary_from_sizes(entries: I) -> DurableMrfBacklogSummary -where - I: IntoIterator, -{ - let mut buckets = HashMap::::new(); - for (bucket_name, entry_size) in entries { +#[derive(Debug, Clone, Default)] +struct DurableMrfBacklogTracker { + available: bool, + buckets: HashMap, +} + +impl DurableMrfBacklogTracker { + fn add_entry(&mut self, bucket_name: String, entry_size: i64) { let Ok(size) = u64::try_from(entry_size) else { - return DurableMrfBacklogSummary::default(); + self.available = false; + self.buckets.clear(); + return; }; - let bucket = match buckets.entry(bucket_name) { + if !self.available { + return; + } + + let bucket = match self.buckets.entry(bucket_name) { Entry::Occupied(entry) => entry.into_mut(), Entry::Vacant(entry) => { let bucket = entry.key().clone(); @@ -118,12 +145,93 @@ where bucket.bytes = bucket.bytes.saturating_add(size); } - DurableMrfBacklogSummary { - available: true, - buckets: buckets.into_values().collect(), + fn into_summary(self) -> DurableMrfBacklogSummary { + if !self.available { + return DurableMrfBacklogSummary::default(); + } + + DurableMrfBacklogSummary { + available: true, + buckets: self.buckets.into_values().collect(), + } } } +#[derive(Debug, Clone, Default)] +struct MrfBacklogObservabilityTracker { + buckets: HashMap, +} + +impl MrfBacklogObservabilityTracker { + fn bucket_mut(&mut self, bucket_name: &str) -> &mut MrfBucketBacklogObservability { + match self.buckets.entry(bucket_name.to_string()) { + Entry::Occupied(entry) => entry.into_mut(), + Entry::Vacant(entry) => entry.insert(MrfBucketBacklogObservability { + bucket: bucket_name.to_string(), + ..Default::default() + }), + } + } + + fn add_pending(&mut self, entry: &MrfReplicateEntry) { + let Ok(size) = u64::try_from(entry.size) else { + return; + }; + let bucket = self.bucket_mut(&entry.bucket); + bucket.pending_count = bucket.pending_count.saturating_add(1); + bucket.pending_bytes = bucket.pending_bytes.saturating_add(size); + } + + fn flush_pending_entries<'a>(&mut self, entries: impl IntoIterator, duration_millis: u64) { + for entry in entries { + let Ok(size) = u64::try_from(entry.size) else { + continue; + }; + let bucket = self.bucket_mut(&entry.bucket); + bucket.pending_count = bucket.pending_count.saturating_sub(1); + bucket.pending_bytes = bucket.pending_bytes.saturating_sub(size); + bucket.last_flush_duration_millis = duration_millis; + } + } + + fn record_drop(&mut self, entry: &MrfReplicateEntry) { + let bucket = self.bucket_mut(&entry.bucket); + bucket.dropped_count = bucket.dropped_count.saturating_add(1); + } + + fn record_missed(&mut self, bucket_name: &str) { + let bucket = self.bucket_mut(bucket_name); + bucket.missed_count = bucket.missed_count.saturating_add(1); + } + + fn record_flush_failure(&mut self, duration_millis: u64) { + for bucket in self.buckets.values_mut().filter(|bucket| bucket.pending_count > 0) { + bucket.flush_failure_count = bucket.flush_failure_count.saturating_add(1); + bucket.last_flush_duration_millis = duration_millis; + } + } + + fn snapshot(&self) -> MrfBacklogObservabilitySummary { + MrfBacklogObservabilitySummary { + buckets: self.buckets.values().cloned().collect(), + } + } +} + +fn durable_mrf_backlog_summary_from_sizes(entries: I) -> DurableMrfBacklogSummary +where + I: IntoIterator, +{ + let mut tracker = DurableMrfBacklogTracker { + available: true, + ..Default::default() + }; + for (bucket_name, entry_size) in entries { + tracker.add_entry(bucket_name, entry_size); + } + tracker.into_summary() +} + fn set_durable_mrf_backlog_summary(summary: DurableMrfBacklogSummary) { match DURABLE_MRF_BACKLOG_SUMMARY.write() { Ok(mut guard) => *guard = summary, @@ -138,6 +246,40 @@ pub fn durable_mrf_backlog_summary_snapshot() -> DurableMrfBacklogSummary { } } +pub fn mrf_backlog_observability_snapshot() -> MrfBacklogObservabilitySummary { + match MRF_BACKLOG_OBSERVABILITY.read() { + Ok(guard) => guard.snapshot(), + Err(poisoned) => poisoned.into_inner().snapshot(), + } +} + +fn update_mrf_backlog_observability(mut update: impl FnMut(&mut MrfBacklogObservabilityTracker)) { + match MRF_BACKLOG_OBSERVABILITY.write() { + Ok(mut guard) => update(&mut guard), + Err(poisoned) => update(&mut poisoned.into_inner()), + } +} + +fn observe_mrf_pending(entry: &MrfReplicateEntry) { + update_mrf_backlog_observability(|tracker| tracker.add_pending(entry)); +} + +fn observe_mrf_pending_flushed(entries: &[MrfReplicateEntry], duration_millis: u64) { + update_mrf_backlog_observability(|tracker| tracker.flush_pending_entries(entries, duration_millis)); +} + +fn observe_mrf_drop(entry: &MrfReplicateEntry) { + update_mrf_backlog_observability(|tracker| tracker.record_drop(entry)); +} + +fn observe_mrf_missed(bucket: &str) { + update_mrf_backlog_observability(|tracker| tracker.record_missed(bucket)); +} + +fn observe_mrf_flush_failure(duration_millis: u64) { + update_mrf_backlog_observability(|tracker| tracker.record_flush_failure(duration_millis)); +} + fn durable_mrf_backlog_from_read(result: Result, EcstoreError>) -> DurableMrfBacklog { match result { Ok(data) => match decode_mrf_file(&data) { @@ -302,26 +444,8 @@ impl ReplicationPool { let handle = tokio::spawn(async move { let mut rx = rx; while let Some(operation) = rx.recv().await { - active_counter.fetch_add(1, Ordering::SeqCst); - - match operation { - ReplicationOperation::Object(obj_info) => { - let bucket = obj_info.bucket.clone(); - let size = obj_info.size; - let delete_marker = obj_info.delete_marker; - let op_type = obj_info.op_type; - replicate_object(*obj_info, storage.clone()).await; - stats.dec_q(&bucket, size, delete_marker, op_type); - } - ReplicationOperation::Delete(del_info) => { - let bucket = del_info.bucket.clone(); - let op_type = del_info.op_type; - replicate_delete(*del_info, storage.clone()).await; - stats.dec_q(&bucket, 0, true, op_type); - } - } - - active_counter.fetch_sub(1, Ordering::SeqCst); + let _active = ActiveWorkerGuard::new(active_counter.clone()); + process_replication_operation(operation, stats.clone(), storage.clone()).await; } }); @@ -379,30 +503,8 @@ impl ReplicationPool { let handle = tokio::spawn(async move { let mut rx = rx; while let Some(operation) = rx.recv().await { - active_counter.fetch_add(1, Ordering::SeqCst); - - match operation { - ReplicationOperation::Object(obj_info) => { - let bucket = obj_info.bucket.clone(); - let size = obj_info.size; - let delete_marker = obj_info.delete_marker; - let op_type = obj_info.op_type; - // Perform actual replication (placeholder) - replicate_object(*obj_info, storage.clone()).await; - - stats.dec_q(&bucket, size, delete_marker, op_type); - } - ReplicationOperation::Delete(del_info) => { - let bucket = del_info.bucket.clone(); - let op_type = del_info.op_type; - // Perform actual delete replication (placeholder) - replicate_delete(*del_info, storage.clone()).await; - - stats.dec_q(&bucket, 0, true, op_type); - } - } - - active_counter.fetch_sub(1, Ordering::SeqCst); + let _active = ActiveWorkerGuard::new(active_counter.clone()); + process_replication_operation(operation, stats.clone(), storage.clone()).await; } }); @@ -448,24 +550,8 @@ impl ReplicationPool { let operation = { mrf_rx.lock().await.recv().await }; let Some(operation) = operation else { break }; - active_counter.fetch_add(1, Ordering::SeqCst); - match operation { - ReplicationOperation::Object(obj_info) => { - let bucket = obj_info.bucket.clone(); - let size = obj_info.size; - let delete_marker = obj_info.delete_marker; - let op_type = obj_info.op_type; - replicate_object(*obj_info, storage.clone()).await; - stats.dec_q(&bucket, size, delete_marker, op_type); - } - ReplicationOperation::Delete(del_info) => { - let bucket = del_info.bucket.clone(); - let op_type = del_info.op_type; - replicate_delete(*del_info, storage.clone()).await; - stats.dec_q(&bucket, 0, true, op_type); - } - } - active_counter.fetch_sub(1, Ordering::SeqCst); + let _active = ActiveWorkerGuard::new(active_counter.clone()); + process_replication_operation(operation, stats.clone(), storage.clone()).await; } }); self.task_handles.lock().await.push(handle); @@ -902,6 +988,10 @@ impl ReplicationPool { // sustained failure storm can't grow it without limit. const MRF_PENDING_CAP: usize = 200_000; let mut pending: Vec = Vec::new(); + let mut durable_tracker = DurableMrfBacklogTracker { + available: true, + ..Default::default() + }; let mut flushed_len = 0usize; let mut dirty = false; let mut capped = false; @@ -915,6 +1005,7 @@ impl ReplicationPool { entry = rx.recv() => match entry { Some(e) => { if pending.len() >= MRF_PENDING_CAP { + observe_mrf_drop(&e); dec_mrf_entries(stats.as_ref(), std::slice::from_ref(&e)); if !capped { capped = true; @@ -927,16 +1018,19 @@ impl ReplicationPool { } continue; } + durable_tracker.add_entry(e.bucket.clone(), e.size); + observe_mrf_pending(&e); pending.push(e); dirty = true; // Flush eagerly once enough new entries have accumulated // since the last write (measured against the flushed // set, not the absolute length, so a large backlog is // not rewritten on every single add). - if pending.len() - flushed_len >= 1000 && flush_mrf_to_disk(&pending, &storage).await { - set_durable_mrf_backlog_summary(durable_mrf_backlog_summary_from_sizes( - pending.iter().map(|entry| (entry.bucket.clone(), entry.size)), - )); + if pending.len() - flushed_len >= 1000 + && let Some(duration_millis) = flush_mrf_to_disk(&pending, &storage).await + { + set_durable_mrf_backlog_summary(durable_tracker.clone().into_summary()); + observe_mrf_pending_flushed(&pending[flushed_len..], duration_millis); dec_mrf_entries(stats.as_ref(), &pending[flushed_len..]); flushed_len = pending.len(); dirty = false; @@ -944,20 +1038,18 @@ impl ReplicationPool { } None => { // Channel closed (pool shutting down) — final flush. - if dirty && flush_mrf_to_disk(&pending, &storage).await { - set_durable_mrf_backlog_summary(durable_mrf_backlog_summary_from_sizes( - pending.iter().map(|entry| (entry.bucket.clone(), entry.size)), - )); + if dirty && let Some(duration_millis) = flush_mrf_to_disk(&pending, &storage).await { + set_durable_mrf_backlog_summary(durable_tracker.clone().into_summary()); + observe_mrf_pending_flushed(&pending[flushed_len..], duration_millis); dec_mrf_entries(stats.as_ref(), &pending[flushed_len..]); } break; } }, _ = interval.tick() => { - if dirty && flush_mrf_to_disk(&pending, &storage).await { - set_durable_mrf_backlog_summary(durable_mrf_backlog_summary_from_sizes( - pending.iter().map(|entry| (entry.bucket.clone(), entry.size)), - )); + if dirty && let Some(duration_millis) = flush_mrf_to_disk(&pending, &storage).await { + set_durable_mrf_backlog_summary(durable_tracker.clone().into_summary()); + observe_mrf_pending_flushed(&pending[flushed_len..], duration_millis); dec_mrf_entries(stats.as_ref(), &pending[flushed_len..]); flushed_len = pending.len(); dirty = false; @@ -977,30 +1069,8 @@ impl ReplicationPool { stats: Arc, ) { while let Some(operation) = rx.recv().await { - active_counter.fetch_add(1, Ordering::SeqCst); - - match operation { - ReplicationOperation::Object(obj_info) => { - let bucket = obj_info.bucket.clone(); - let size = obj_info.size; - let delete_marker = obj_info.delete_marker; - let op_type = obj_info.op_type; - // Perform actual replication (placeholder) - replicate_object(*obj_info, self.storage.clone()).await; - - stats.dec_q(&bucket, size, delete_marker, op_type); - } - ReplicationOperation::Delete(del_info) => { - let bucket = del_info.bucket.clone(); - let op_type = del_info.op_type; - // Perform actual delete replication (placeholder) - replicate_delete(*del_info, self.storage.clone()).await; - - stats.dec_q(&bucket, 0, true, op_type); - } - } - - active_counter.fetch_sub(1, Ordering::SeqCst); + let _active = ActiveWorkerGuard::new(active_counter.clone()); + process_replication_operation(operation, stats.clone(), self.storage.clone()).await; } } @@ -1013,26 +1083,8 @@ impl ReplicationPool { storage: Arc, ) { while let Some(operation) = rx.recv().await { - active_counter.fetch_add(1, Ordering::SeqCst); - - match operation { - ReplicationOperation::Object(obj_info) => { - let bucket = obj_info.bucket.clone(); - let size = obj_info.size; - let delete_marker = obj_info.delete_marker; - let op_type = obj_info.op_type; - replicate_object(*obj_info, storage.clone()).await; - stats.dec_q(&bucket, size, delete_marker, op_type); - } - ReplicationOperation::Delete(del_info) => { - let bucket = del_info.bucket.clone(); - let op_type = del_info.op_type; - replicate_delete(*del_info, storage.clone()).await; - stats.dec_q(&bucket, 0, true, op_type); - } - } - - active_counter.fetch_sub(1, Ordering::SeqCst); + let _active = ActiveWorkerGuard::new(active_counter.clone()); + process_replication_operation(operation, stats.clone(), storage.clone()).await; } } @@ -1044,27 +1096,8 @@ impl ReplicationPool { stats: Arc, ) { while let Some(operation) = rx.recv().await { - active_counter.fetch_add(1, Ordering::SeqCst); - - match operation { - ReplicationOperation::Object(obj_info) => { - let bucket = obj_info.bucket.clone(); - let size = obj_info.size; - let delete_marker = obj_info.delete_marker; - let op_type = obj_info.op_type; - replicate_object(obj_info.as_ref().clone(), self.storage.clone()).await; - - stats.dec_q(&bucket, size, delete_marker, op_type); - } - ReplicationOperation::Delete(del_info) => { - let bucket = del_info.bucket.clone(); - let op_type = del_info.op_type; - replicate_delete(*del_info, self.storage.clone()).await; - stats.dec_q(&bucket, 0, true, op_type); - } - } - - active_counter.fetch_sub(1, Ordering::SeqCst); + let _active = ActiveWorkerGuard::new(active_counter.clone()); + process_replication_operation(operation, stats.clone(), self.storage.clone()).await; } } @@ -1395,6 +1428,76 @@ impl ReplicationPool { } } +struct ActiveWorkerGuard { + counter: Arc, +} + +impl ActiveWorkerGuard { + fn new(counter: Arc) -> Self { + counter.fetch_add(1, Ordering::SeqCst); + Self { counter } + } +} + +impl Drop for ActiveWorkerGuard { + fn drop(&mut self) { + self.counter.fetch_sub(1, Ordering::SeqCst); + } +} + +struct ReplicationBacklogGuard { + stats: Arc, + bucket: String, + size: i64, + is_delete_marker: bool, + op_type: ReplicationType, +} + +impl ReplicationBacklogGuard { + fn for_object(stats: Arc, object: &ReplicateObjectInfo) -> Self { + Self { + stats, + bucket: object.bucket.clone(), + size: object.size, + is_delete_marker: object.delete_marker, + op_type: object.op_type, + } + } + + fn for_delete(stats: Arc, delete: &DeletedObjectReplicationInfo) -> Self { + Self { + stats, + bucket: delete.bucket.clone(), + size: 0, + is_delete_marker: true, + op_type: delete.op_type, + } + } +} + +impl Drop for ReplicationBacklogGuard { + fn drop(&mut self) { + self.stats.dec_q(&self.bucket, self.size, self.is_delete_marker, self.op_type); + } +} + +async fn process_replication_operation( + operation: ReplicationOperation, + stats: Arc, + storage: Arc, +) { + match operation { + ReplicationOperation::Object(obj_info) => { + let _backlog = ReplicationBacklogGuard::for_object(stats, obj_info.as_ref()); + replicate_object(*obj_info, storage).await; + } + ReplicationOperation::Delete(del_info) => { + let _backlog = ReplicationBacklogGuard::for_delete(stats, del_info.as_ref()); + replicate_delete(*del_info, storage).await; + } + } +} + async fn queue_mrf_save_entry( tx: &Sender, entry: MrfReplicateEntry, @@ -1414,6 +1517,7 @@ async fn queue_mrf_save_entry( queue_type = queue_type, "MRF save channel unavailable — replication failure entry could not be persisted for retry" ); + observe_mrf_missed(&entry.bucket); ReplicationQueueAdmission::Missed } @@ -1424,15 +1528,18 @@ fn dec_mrf_entries(stats: &ReplicationStats, entries: &[MrfReplicateEntry]) { } /// Encodes `entries` and overwrites the MRF persistence file. -/// Returns `true` on success; on failure logs the error and returns `false`. -/// Callers must NOT clear their in-memory buffer on `false` so the next tick +/// Returns the flush duration on success; on failure logs the error and returns `None`. +/// Callers must NOT clear their in-memory buffer on `None` so the next tick /// can retry — otherwise a transient storage error permanently drops the batch. -async fn flush_mrf_to_disk(entries: &[MrfReplicateEntry], storage: &Arc) -> bool { +async fn flush_mrf_to_disk(entries: &[MrfReplicateEntry], storage: &Arc) -> Option { + let started = Instant::now(); match encode_mrf_file(entries) { Ok(data) => { if let Err(e) = ReplicationConfigStore::save(storage.clone(), ReplicationMetadataStore::MRF_REPLICATION_FILE, data).await { + let duration_millis = duration_millis_u64(started.elapsed()); + observe_mrf_flush_failure(duration_millis); warn!( component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_REPLICATION, @@ -1440,11 +1547,12 @@ async fn flush_mrf_to_disk(entries: &[MrfReplicateEntry] error = %e, "Failed to flush MRF entries to disk" ); - return false; + return None; } - true + Some(duration_millis_u64(started.elapsed())) } Err(e) => { + observe_mrf_flush_failure(0); warn!( component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_REPLICATION, @@ -1452,11 +1560,15 @@ async fn flush_mrf_to_disk(entries: &[MrfReplicateEntry] error = %e, "Failed to encode MRF entries for disk flush" ); - false + None } } } +fn duration_millis_u64(duration: std::time::Duration) -> u64 { + u64::try_from(duration.as_millis()).unwrap_or(u64::MAX) +} + /// Load bucket resync metadata from disk async fn load_bucket_resync_metadata( bucket: &str, @@ -2664,6 +2776,132 @@ mod tests { assert_eq!(received.object, "second"); } + #[tokio::test] + async fn mrf_save_admission_records_missed_when_channel_is_closed() { + let (tx, rx) = mpsc::channel(1); + drop(rx); + let bucket = "mrf-missed-hook-bucket"; + + let admission = queue_mrf_save_entry( + &tx, + MrfReplicateEntry { + bucket: bucket.to_string(), + object: "missed".to_string(), + version_id: None, + retry_count: 1, + size: 1, + op: MrfOpKind::Object, + delete_marker_version_id: None, + delete_marker: false, + delete_marker_mtime: None, + }, + "test", + ) + .await; + + assert_eq!(admission, ReplicationQueueAdmission::Missed); + let snapshot = mrf_backlog_observability_snapshot(); + let bucket = snapshot + .buckets + .iter() + .find(|stats| stats.bucket == "mrf-missed-hook-bucket") + .expect("missed MRF admission should be observable"); + assert_eq!(bucket.missed_count, 1); + } + + #[tokio::test] + async fn mrf_flush_failure_keeps_pending_backlog_observable() { + let shared = empty_resync_shared_state(); + shared.fail_next_write.store(true, Ordering::SeqCst); + let storage = Arc::new(LoadResyncNodeStore::new("mrf-flush-failure", shared)); + let entry = MrfReplicateEntry { + bucket: "mrf-flush-failure-bucket".to_string(), + object: "pending".to_string(), + version_id: None, + retry_count: 1, + size: 2048, + op: MrfOpKind::Object, + delete_marker_version_id: None, + delete_marker: false, + delete_marker_mtime: None, + }; + observe_mrf_pending(&entry); + + let result = flush_mrf_to_disk(std::slice::from_ref(&entry), &storage).await; + + assert_eq!(result, None); + let snapshot = mrf_backlog_observability_snapshot(); + let bucket = snapshot + .buckets + .iter() + .find(|stats| stats.bucket == "mrf-flush-failure-bucket") + .expect("failed MRF flush should keep the bucket observable"); + assert_eq!(bucket.pending_count, 1); + assert_eq!(bucket.pending_bytes, 2048); + assert_eq!(bucket.flush_failure_count, 1); + } + + #[tokio::test] + async fn replication_backlog_guard_decrements_on_drop() { + let stats = Arc::new(ReplicationStats::new()); + stats.inc_q("guard-bucket", 256, false, ReplicationType::Object); + + { + let object = ReplicateObjectInfo { + bucket: "guard-bucket".to_string(), + size: 256, + op_type: ReplicationType::Object, + ..Default::default() + }; + let _guard = ReplicationBacklogGuard::for_object(stats.clone(), &object); + } + + let queued = stats.get_latest_replication_stats("guard-bucket").await; + assert_eq!(queued.replication_stats.q_stat.curr.count, 0); + assert_eq!(queued.replication_stats.q_stat.curr.bytes, 0); + } + + #[test] + fn mrf_observability_tracker_separates_pending_drop_miss_and_flush_failure() { + let first = MrfReplicateEntry { + bucket: "tracker-bucket".to_string(), + object: "first".to_string(), + version_id: None, + retry_count: 1, + size: 1024, + op: MrfOpKind::Object, + delete_marker_version_id: None, + delete_marker: false, + delete_marker_mtime: None, + }; + let second = MrfReplicateEntry { + object: "second".to_string(), + size: 512, + ..first.clone() + }; + let mut tracker = MrfBacklogObservabilityTracker::default(); + + tracker.add_pending(&first); + tracker.add_pending(&second); + tracker.record_drop(&second); + tracker.record_missed("tracker-bucket"); + tracker.record_flush_failure(7); + tracker.flush_pending_entries([&first], 11); + + let snapshot = tracker.snapshot(); + let bucket = snapshot + .buckets + .iter() + .find(|stats| stats.bucket == "tracker-bucket") + .expect("tracker bucket should be present"); + assert_eq!(bucket.pending_count, 1); + assert_eq!(bucket.pending_bytes, 512); + assert_eq!(bucket.dropped_count, 1); + assert_eq!(bucket.missed_count, 1); + assert_eq!(bucket.flush_failure_count, 1); + assert_eq!(bucket.last_flush_duration_millis, 11); + } + #[test] fn auto_resume_resync_only_for_inflight_states() { assert!(should_auto_resume_resync(ResyncStatusType::ResyncPending)); diff --git a/crates/obs/src/metrics/collectors/bucket_replication.rs b/crates/obs/src/metrics/collectors/bucket_replication.rs index 32d7151fd..23891bc8e 100644 --- a/crates/obs/src/metrics/collectors/bucket_replication.rs +++ b/crates/obs/src/metrics/collectors/bucket_replication.rs @@ -20,6 +20,8 @@ use crate::metrics::schema::bucket_replication::{ BUCKET_REPL_CURRENT_BACKLOG_COUNT_MD, BUCKET_REPL_DURABLE_MRF_AVAILABLE_MD, BUCKET_REPL_DURABLE_MRF_BACKLOG_BYTES_MD, BUCKET_REPL_DURABLE_MRF_BACKLOG_COUNT_MD, BUCKET_REPL_LAST_HR_FAILED_BYTES_MD, BUCKET_REPL_LAST_HR_FAILED_COUNT_MD, BUCKET_REPL_LAST_MIN_FAILED_BYTES_MD, BUCKET_REPL_LAST_MIN_FAILED_COUNT_MD, BUCKET_REPL_LATENCY_MS_MD, + BUCKET_REPL_MRF_DROPPED_COUNT_MD, BUCKET_REPL_MRF_FLUSH_FAILURES_MD, BUCKET_REPL_MRF_LAST_FLUSH_DURATION_MILLIS_MD, + BUCKET_REPL_MRF_MISSED_COUNT_MD, BUCKET_REPL_MRF_PENDING_BYTES_MD, BUCKET_REPL_MRF_PENDING_COUNT_MD, BUCKET_REPL_PROXIED_DELETE_TAGGING_REQUESTS_FAILURES_MD, BUCKET_REPL_PROXIED_DELETE_TAGGING_REQUESTS_TOTAL_MD, BUCKET_REPL_PROXIED_GET_REQUESTS_FAILURES_MD, BUCKET_REPL_PROXIED_GET_REQUESTS_TOTAL_MD, BUCKET_REPL_PROXIED_GET_TAGGING_REQUESTS_FAILURES_MD, BUCKET_REPL_PROXIED_GET_TAGGING_REQUESTS_TOTAL_MD, @@ -34,7 +36,7 @@ use crate::metrics::schema::bucket_replication::{ use std::borrow::Cow; const BASE_BUCKET_REPLICATION_METRICS_PER_BUCKET: usize = 25; -const BASE_BUCKET_REPLICATION_BACKLOG_METRICS_PER_BUCKET: usize = 5; +const BASE_BUCKET_REPLICATION_BACKLOG_METRICS_PER_BUCKET: usize = 11; #[derive(Debug, Clone, Default)] pub struct BucketReplicationTargetStats { @@ -91,6 +93,12 @@ pub(crate) struct BucketReplicationBacklogStats { pub(crate) durable_mrf_available: bool, pub(crate) durable_mrf_backlog_count: u64, pub(crate) durable_mrf_backlog_bytes: u64, + pub(crate) mrf_pending_count: u64, + pub(crate) mrf_pending_bytes: u64, + pub(crate) mrf_dropped_count: u64, + pub(crate) mrf_missed_count: u64, + pub(crate) mrf_flush_failures: u64, + pub(crate) mrf_last_flush_duration_millis: u64, } pub fn collect_bucket_replication_bandwidth_metrics(stats: &[BucketReplicationBandwidthStats]) -> Vec { @@ -309,6 +317,34 @@ pub(crate) fn collect_bucket_replication_backlog_metrics(stats: &[BucketReplicat PrometheusMetric::from_descriptor(&BUCKET_REPL_DURABLE_MRF_BACKLOG_BYTES_MD, stat.durable_mrf_backlog_bytes as f64) .with_label(BUCKET_L, bucket_label), ); + let bucket_label: Cow<'static, str> = Cow::Owned(stat.bucket.clone()); + metrics.push( + PrometheusMetric::from_descriptor(&BUCKET_REPL_MRF_PENDING_COUNT_MD, stat.mrf_pending_count as f64) + .with_label(BUCKET_L, bucket_label.clone()), + ); + metrics.push( + PrometheusMetric::from_descriptor(&BUCKET_REPL_MRF_PENDING_BYTES_MD, stat.mrf_pending_bytes as f64) + .with_label(BUCKET_L, bucket_label.clone()), + ); + metrics.push( + PrometheusMetric::from_descriptor(&BUCKET_REPL_MRF_DROPPED_COUNT_MD, stat.mrf_dropped_count as f64) + .with_label(BUCKET_L, bucket_label.clone()), + ); + metrics.push( + PrometheusMetric::from_descriptor(&BUCKET_REPL_MRF_MISSED_COUNT_MD, stat.mrf_missed_count as f64) + .with_label(BUCKET_L, bucket_label.clone()), + ); + metrics.push( + PrometheusMetric::from_descriptor(&BUCKET_REPL_MRF_FLUSH_FAILURES_MD, stat.mrf_flush_failures as f64) + .with_label(BUCKET_L, bucket_label.clone()), + ); + metrics.push( + PrometheusMetric::from_descriptor( + &BUCKET_REPL_MRF_LAST_FLUSH_DURATION_MILLIS_MD, + stat.mrf_last_flush_duration_millis as f64, + ) + .with_label(BUCKET_L, bucket_label), + ); } metrics @@ -427,10 +463,16 @@ mod tests { durable_mrf_available: true, durable_mrf_backlog_count: 2, durable_mrf_backlog_bytes: 2048, + mrf_pending_count: 1, + mrf_pending_bytes: 512, + mrf_dropped_count: 3, + mrf_missed_count: 4, + mrf_flush_failures: 5, + mrf_last_flush_duration_millis: 6, }]; let metrics = collect_bucket_replication_backlog_metrics(&stats); - assert_eq!(metrics.len(), 5); + assert_eq!(metrics.len(), 11); let backlog_count_name = BUCKET_REPL_CURRENT_BACKLOG_COUNT_MD.get_full_metric_name(); assert!(metrics.iter().any(|metric| { @@ -466,6 +508,48 @@ mod tests { && metric.value == 2048.0 && metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1") })); + + let pending_count_name = BUCKET_REPL_MRF_PENDING_COUNT_MD.get_full_metric_name(); + assert!(metrics.iter().any(|metric| { + metric.name == pending_count_name + && metric.value == 1.0 + && metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1") + })); + + let pending_bytes_name = BUCKET_REPL_MRF_PENDING_BYTES_MD.get_full_metric_name(); + assert!(metrics.iter().any(|metric| { + metric.name == pending_bytes_name + && metric.value == 512.0 + && metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1") + })); + + let dropped_count_name = BUCKET_REPL_MRF_DROPPED_COUNT_MD.get_full_metric_name(); + assert!(metrics.iter().any(|metric| { + metric.name == dropped_count_name + && metric.value == 3.0 + && metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1") + })); + + let missed_count_name = BUCKET_REPL_MRF_MISSED_COUNT_MD.get_full_metric_name(); + assert!(metrics.iter().any(|metric| { + metric.name == missed_count_name + && metric.value == 4.0 + && metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1") + })); + + let flush_failures_name = BUCKET_REPL_MRF_FLUSH_FAILURES_MD.get_full_metric_name(); + assert!(metrics.iter().any(|metric| { + metric.name == flush_failures_name + && metric.value == 5.0 + && metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1") + })); + + let flush_duration_name = BUCKET_REPL_MRF_LAST_FLUSH_DURATION_MILLIS_MD.get_full_metric_name(); + assert!(metrics.iter().any(|metric| { + metric.name == flush_duration_name + && metric.value == 6.0 + && metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1") + })); } #[test] @@ -484,6 +568,12 @@ mod tests { durable_mrf_available: true, durable_mrf_backlog_count: 2, durable_mrf_backlog_bytes: 4096, + mrf_pending_count: 1, + mrf_pending_bytes: 2, + mrf_dropped_count: 3, + mrf_missed_count: 4, + mrf_flush_failures: 5, + mrf_last_flush_duration_millis: 6, }]; let metrics = collect_bucket_replication_backlog_metrics(&stats); @@ -493,6 +583,12 @@ mod tests { BUCKET_REPL_DURABLE_MRF_AVAILABLE_MD.get_full_metric_name(), BUCKET_REPL_DURABLE_MRF_BACKLOG_COUNT_MD.get_full_metric_name(), BUCKET_REPL_DURABLE_MRF_BACKLOG_BYTES_MD.get_full_metric_name(), + BUCKET_REPL_MRF_PENDING_COUNT_MD.get_full_metric_name(), + BUCKET_REPL_MRF_PENDING_BYTES_MD.get_full_metric_name(), + BUCKET_REPL_MRF_DROPPED_COUNT_MD.get_full_metric_name(), + BUCKET_REPL_MRF_MISSED_COUNT_MD.get_full_metric_name(), + BUCKET_REPL_MRF_FLUSH_FAILURES_MD.get_full_metric_name(), + BUCKET_REPL_MRF_LAST_FLUSH_DURATION_MILLIS_MD.get_full_metric_name(), ]; for name in backlog_names { diff --git a/crates/obs/src/metrics/schema/bucket_replication.rs b/crates/obs/src/metrics/schema/bucket_replication.rs index e029774ef..28237ecdf 100644 --- a/crates/obs/src/metrics/schema/bucket_replication.rs +++ b/crates/obs/src/metrics/schema/bucket_replication.rs @@ -38,6 +38,12 @@ const CURRENT_BACKLOG_BYTES: &str = "current_backlog_bytes"; const DURABLE_MRF_AVAILABLE: &str = "durable_mrf_available"; const DURABLE_MRF_BACKLOG_COUNT: &str = "durable_mrf_backlog_count"; const DURABLE_MRF_BACKLOG_BYTES: &str = "durable_mrf_backlog_bytes"; +const MRF_PENDING_COUNT: &str = "mrf_pending_count"; +const MRF_PENDING_BYTES: &str = "mrf_pending_bytes"; +const MRF_DROPPED_COUNT: &str = "mrf_dropped_count"; +const MRF_MISSED_COUNT: &str = "mrf_missed_count"; +const MRF_FLUSH_FAILURES: &str = "mrf_flush_failures"; +const MRF_LAST_FLUSH_DURATION_MILLIS: &str = "mrf_last_flush_duration_millis"; pub static BUCKET_REPL_LAST_HR_FAILED_BYTES_MD: LazyLock = LazyLock::new(|| { new_gauge_md( @@ -129,6 +135,60 @@ pub static BUCKET_REPL_DURABLE_MRF_BACKLOG_BYTES_MD: LazyLock ) }); +pub static BUCKET_REPL_MRF_PENDING_COUNT_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::from(MRF_PENDING_COUNT), + "Current number of MRF entries waiting to be flushed to the durable recovery file for a bucket on this node", + &[BUCKET_L], + subsystems::BUCKET_REPLICATION, + ) +}); + +pub static BUCKET_REPL_MRF_PENDING_BYTES_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::from(MRF_PENDING_BYTES), + "Current bytes represented by MRF entries waiting to be flushed to the durable recovery file for a bucket on this node", + &[BUCKET_L], + subsystems::BUCKET_REPLICATION, + ) +}); + +pub static BUCKET_REPL_MRF_DROPPED_COUNT_MD: LazyLock = LazyLock::new(|| { + new_counter_md( + MetricName::from(MRF_DROPPED_COUNT), + "Total number of MRF entries dropped after the bounded pending backlog cap was reached for a bucket on this node", + &[BUCKET_L], + subsystems::BUCKET_REPLICATION, + ) +}); + +pub static BUCKET_REPL_MRF_MISSED_COUNT_MD: LazyLock = LazyLock::new(|| { + new_counter_md( + MetricName::from(MRF_MISSED_COUNT), + "Total number of MRF entries that could not be admitted to the save channel for a bucket on this node", + &[BUCKET_L], + subsystems::BUCKET_REPLICATION, + ) +}); + +pub static BUCKET_REPL_MRF_FLUSH_FAILURES_MD: LazyLock = LazyLock::new(|| { + new_counter_md( + MetricName::from(MRF_FLUSH_FAILURES), + "Total number of durable MRF flush failures observed while a bucket had pending MRF entries on this node", + &[BUCKET_L], + subsystems::BUCKET_REPLICATION, + ) +}); + +pub static BUCKET_REPL_MRF_LAST_FLUSH_DURATION_MILLIS_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::from(MRF_LAST_FLUSH_DURATION_MILLIS), + "Duration in milliseconds of the last durable MRF flush that touched pending entries for a bucket on this node", + &[BUCKET_L], + subsystems::BUCKET_REPLICATION, + ) +}); + pub static BUCKET_REPL_PROXIED_DELETE_TAGGING_REQUESTS_TOTAL_MD: LazyLock = LazyLock::new(|| { new_counter_md( MetricName::ProxiedDeleteTaggingRequestsTotal, diff --git a/crates/obs/src/metrics/stats_collector.rs b/crates/obs/src/metrics/stats_collector.rs index f9040cc3c..170279e37 100644 --- a/crates/obs/src/metrics/stats_collector.rs +++ b/crates/obs/src/metrics/stats_collector.rs @@ -206,6 +206,12 @@ async fn obs_bucket_replication_stats_bundle() -> (Vec, durable_mrf_available: stats.durable_mrf_available, durable_mrf_backlog_count: stats.durable_mrf_backlog_count, durable_mrf_backlog_bytes: stats.durable_mrf_backlog_bytes, + mrf_pending_count: stats.mrf_pending_count, + mrf_pending_bytes: stats.mrf_pending_bytes, + mrf_dropped_count: stats.mrf_dropped_count, + mrf_missed_count: stats.mrf_missed_count, + mrf_flush_failures: stats.mrf_flush_failures, + mrf_last_flush_duration_millis: stats.mrf_last_flush_duration_millis, }); detail_stats.push(bucket_replication_detail_from_snapshot(stats)); } diff --git a/crates/obs/src/metrics/storage_api.rs b/crates/obs/src/metrics/storage_api.rs index 55221d7aa..e1286ebe1 100644 --- a/crates/obs/src/metrics/storage_api.rs +++ b/crates/obs/src/metrics/storage_api.rs @@ -18,7 +18,8 @@ use std::time::Duration; pub(crate) use rustfs_ecstore::api::bucket::bandwidth::monitor::Monitor as ObsBucketBandwidthMonitor; pub(crate) use rustfs_ecstore::api::bucket::metadata_sys::get_quota_config as obs_get_quota_config; use rustfs_ecstore::api::bucket::replication::{ - DurableMrfBucketBacklog, durable_mrf_backlog_summary_snapshot, get_global_replication_stats, + DurableMrfBucketBacklog, MrfBucketBacklogObservability, durable_mrf_backlog_summary_snapshot, get_global_replication_stats, + mrf_backlog_observability_snapshot, }; pub(crate) use rustfs_ecstore::api::capacity::{ get_total_usable_capacity as obs_get_total_usable_capacity, @@ -76,6 +77,12 @@ pub(crate) struct ObsBucketReplicationStatsSnapshot { pub(crate) durable_mrf_available: bool, pub(crate) durable_mrf_backlog_count: u64, pub(crate) durable_mrf_backlog_bytes: u64, + pub(crate) mrf_pending_count: u64, + pub(crate) mrf_pending_bytes: u64, + pub(crate) mrf_dropped_count: u64, + pub(crate) mrf_missed_count: u64, + pub(crate) mrf_flush_failures: u64, + pub(crate) mrf_last_flush_duration_millis: u64, pub(crate) targets: Vec, } @@ -152,6 +159,7 @@ fn bucket_replication_stats_snapshot_from_parts( proxy: ObsBucketReplicationProxySnapshot, durable_mrf_available: bool, durable_bucket: DurableMrfBucketBacklog, + mrf_observability: MrfBucketBacklogObservability, ) -> ObsBucketReplicationStatsSnapshot { ObsBucketReplicationStatsSnapshot { bucket, @@ -185,6 +193,12 @@ fn bucket_replication_stats_snapshot_from_parts( durable_mrf_available, durable_mrf_backlog_count: durable_bucket.count, durable_mrf_backlog_bytes: durable_bucket.bytes, + mrf_pending_count: mrf_observability.pending_count, + mrf_pending_bytes: mrf_observability.pending_bytes, + mrf_dropped_count: mrf_observability.dropped_count, + mrf_missed_count: mrf_observability.missed_count, + mrf_flush_failures: mrf_observability.flush_failure_count, + mrf_last_flush_duration_millis: mrf_observability.last_flush_duration_millis, targets: runtime.targets, } } @@ -196,7 +210,8 @@ pub(crate) async fn obs_bucket_replication_stats_snapshot() -> Vec Vec>(); - let mut bucket_names = Vec::with_capacity(all_bucket_stats.len().saturating_add(durable_buckets.len())); + let mrf_observability = if replication_storage_available { + mrf_backlog_observability_snapshot() + } else { + Default::default() + }; + let mrf_observability_buckets = mrf_observability + .buckets + .into_iter() + .map(|bucket| (bucket.bucket.clone(), bucket)) + .collect::>(); + let mut bucket_names = Vec::with_capacity( + all_bucket_stats + .len() + .saturating_add(durable_buckets.len()) + .saturating_add(mrf_observability_buckets.len()), + ); bucket_names.extend(all_bucket_stats.keys().cloned()); bucket_names.extend( durable_buckets @@ -215,6 +245,12 @@ pub(crate) async fn obs_bucket_replication_stats_snapshot() -> Vec Vec