diff --git a/Cargo.toml b/Cargo.toml index 377b0d98..b12aadf0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -35,7 +35,7 @@ clap_complete = "4.5" # Serialization serde = { version = "1.0", features = ["derive"] } -serde_json = "1.0" +serde_json = { version = "1.0", features = ["raw_value"] } toml = "0.9" quick-xml = { version = "0.41", features = ["serialize"] } diff --git a/crates/cli/src/commands/replicate.rs b/crates/cli/src/commands/replicate.rs index f6836172..c9031a0c 100644 --- a/crates/cli/src/commands/replicate.rs +++ b/crates/cli/src/commands/replicate.rs @@ -3777,6 +3777,44 @@ mod tests { assert!(value["data"].get("healthy").is_none()); } + #[test] + fn status_output_reports_captured_minio_wire_without_fabricating_latency() { + let metrics: ReplicationMetrics = serde_json::from_str(include_str!( + "../../../core/tests/fixtures/replication_metrics_minio_v1.json" + )) + .expect("captured MinIO-compatible metrics"); + let value = serde_json::to_value(replication_status_output("source", metrics.clone())) + .expect("status JSON"); + + assert_eq!(value["schema_version"], 3); + assert_eq!(value["data"]["availability"], "available"); + assert_eq!(value["data"]["cluster"]["state"], "complete"); + assert_eq!(value["data"]["cluster"]["observed_nodes"], 1); + assert_eq!(value["data"]["cluster"]["expected_nodes"], 1); + assert_eq!(value["data"]["totals"]["replicated_count"], 1); + assert_eq!(value["data"]["totals"]["replicated_size_bytes"], 20); + assert_eq!(value["data"]["targets"][0]["replicated_count"], 1); + assert_eq!(value["data"]["targets"][0]["replicated_size_bytes"], 20); + assert_eq!(value["data"]["targets"][0]["latency"]["average_ms"], 0.0); + assert_eq!( + value["data"]["targets"][0]["latency"]["scope"], + "unavailable" + ); + assert_eq!( + value["data"]["targets"][0]["bandwidth"]["scope"], + "legacy_unknown" + ); + + let formatter = Formatter::new(OutputConfig { + no_color: true, + ..Default::default() + }); + let output = replication_status_lines("source", &metrics, &formatter).join("\n"); + assert!(output.contains("provider=available, cluster=complete (1/1 nodes)")); + assert!(output.contains("Totals: replicated 1 / 0 objects, 20 / 0 bytes")); + assert!(output.contains("1 / 20 bytes 0 / 0 bytes unavailable legacy_unknown")); + } + #[test] fn status_output_marks_legacy_availability_unknown() { let metrics: ReplicationMetrics = serde_json::from_str( diff --git a/crates/cli/tests/replication_metrics.rs b/crates/cli/tests/replication_metrics.rs new file mode 100644 index 00000000..5c83824b --- /dev/null +++ b/crates/cli/tests/replication_metrics.rs @@ -0,0 +1,109 @@ +//! Process-level contracts for MinIO-compatible replication metrics. + +#![cfg(not(windows))] + +mod admin_support; + +use std::process::{Command, Output}; +use std::time::Duration; + +use admin_support::{rc_binary, rc_host_alias, start_admin_test_server}; + +const MINIO_METRICS: &str = + include_str!("../../core/tests/fixtures/replication_metrics_minio_v1.json"); + +fn run_status(json: bool) -> (Output, admin_support::CapturedAdminRequest) { + let config_dir = tempfile::tempdir().expect("create isolated config dir"); + let (endpoint, receiver, handle) = start_admin_test_server(MINIO_METRICS); + let mut command = Command::new(rc_binary()); + command.arg("--no-color"); + if json { + command.arg("--json"); + } else { + command.args(["--format", "human"]); + } + let output = command + .args(["bucket", "replication", "status", "myalias/source-bucket"]) + .env("RC_CONFIG_DIR", config_dir.path()) + .env("RC_HOST_myalias", rc_host_alias(&endpoint)) + .output() + .expect("execute replication status"); + let request = receiver + .recv_timeout(Duration::from_secs(5)) + .expect("captured replication metrics request"); + handle.join().expect("admin test server finished"); + (output, request) +} + +#[test] +fn replication_status_human_accepts_minio_metrics() { + let (output, request) = run_status(false); + + assert!( + output.status.success(), + "stderr: {}", + String::from_utf8_lossy(&output.stderr) + ); + assert!(output.stderr.is_empty()); + let stdout = String::from_utf8(output.stdout).expect("UTF-8 human output"); + assert!( + stdout.contains("provider=available, cluster=complete (1/1 nodes)"), + "stdout: {stdout}" + ); + assert!( + stdout.contains("Totals: replicated 1 / 0 objects, 20 / 0 bytes"), + "stdout: {stdout}" + ); + assert!( + stdout.contains("1 / 20 bytes 0 / 0 bytes unavailable legacy_unknown"), + "stdout: {stdout}" + ); + assert_eq!( + request.target, + "/rustfs/admin/v3/replicationmetrics?bucket=source-bucket" + ); +} + +#[test] +fn replication_status_json_accepts_minio_metrics() { + let (output, request) = run_status(true); + + assert!( + output.status.success(), + "stderr: {}", + String::from_utf8_lossy(&output.stderr) + ); + assert!(output.stderr.is_empty()); + let value: serde_json::Value = + serde_json::from_slice(&output.stdout).expect("v3 replication JSON"); + let schema: serde_json::Value = + serde_json::from_str(include_str!("../../../schemas/output_v3.json")) + .expect("output-v3 schema"); + let validator = jsonschema::validator_for(&schema).expect("compiled output-v3 schema"); + let errors = validator + .iter_errors(&value) + .map(|error| error.to_string()) + .collect::>(); + assert!( + errors.is_empty(), + "replication status violates output-v3 schema: {}", + errors.join("; ") + ); + assert_eq!(value["schema_version"], 3); + assert_eq!(value["type"], "replication"); + assert_eq!(value["status"], "success"); + assert_eq!(value["data"]["availability"], "available"); + assert_eq!(value["data"]["cluster"]["state"], "complete"); + assert_eq!(value["data"]["totals"]["replicated_count"], 1); + assert_eq!(value["data"]["totals"]["replicated_size_bytes"], 20); + assert_eq!(value["data"]["targets"][0]["replicated_count"], 1); + assert_eq!(value["data"]["targets"][0]["replicated_size_bytes"], 20); + assert_eq!( + value["data"]["targets"][0]["latency"]["scope"], + "unavailable" + ); + assert_eq!( + request.target, + "/rustfs/admin/v3/replicationmetrics?bucket=source-bucket" + ); +} diff --git a/crates/core/src/admin/replication.rs b/crates/core/src/admin/replication.rs index bc72f2cd..f27041c7 100644 --- a/crates/core/src/admin/replication.rs +++ b/crates/core/src/admin/replication.rs @@ -117,7 +117,7 @@ pub struct ReplicationTargetMetric { pub extra: BTreeMap, } -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Serialize)] pub struct ReplicationMetrics { pub stats: BTreeMap, pub replica_size: u64, @@ -139,6 +139,490 @@ pub struct ReplicationMetrics { pub extra: BTreeMap, } +#[derive(Debug, Default)] +enum WireField { + #[default] + Missing, + Present(T), +} + +impl WireField { + fn is_present(&self) -> bool { + matches!(self, Self::Present(_)) + } + + fn require(self, name: &str) -> std::result::Result { + match self { + Self::Present(value) => Ok(value), + Self::Missing => Err(format!("missing field `{name}`")), + } + } +} + +impl<'de, T> Deserialize<'de> for WireField +where + T: Deserialize<'de>, +{ + fn deserialize(deserializer: D) -> std::result::Result + where + D: serde::Deserializer<'de>, + { + T::deserialize(deserializer).map(Self::Present) + } +} + +#[derive(Debug, Clone, Copy)] +struct WireCounter(u64); + +impl<'de> Deserialize<'de> for WireCounter { + fn deserialize(deserializer: D) -> std::result::Result + where + D: serde::Deserializer<'de>, + { + // Keep the original token so decimal counters cannot be silently rounded + // through f64 before their integer semantics are validated. + let raw = <&serde_json::value::RawValue>::deserialize(deserializer)?; + parse_wire_counter(raw.get()) + .map(WireCounter) + .map_err(serde::de::Error::custom) + } +} + +fn parse_wire_counter(raw: &str) -> std::result::Result { + let bytes = raw.as_bytes(); + if bytes.is_empty() || bytes[0] == b'-' { + return Err("counter cannot be negative"); + } + if !bytes[0].is_ascii_digit() { + return Err("counter must be a JSON number"); + } + + let exponent_start = bytes + .iter() + .position(|byte| matches!(byte, b'e' | b'E')) + .unwrap_or(bytes.len()); + let mantissa = &bytes[..exponent_start]; + let decimal = mantissa.iter().position(|byte| *byte == b'.'); + let (integer_digits, fraction_digits) = match decimal { + Some(index) => (&mantissa[..index], &mantissa[index + 1..]), + None => (mantissa, &[][..]), + }; + if integer_digits.is_empty() + || !integer_digits.iter().all(u8::is_ascii_digit) + || !fraction_digits.iter().all(u8::is_ascii_digit) + { + return Err("counter must be a JSON number"); + } + + let digits = integer_digits.iter().chain(fraction_digits); + if digits.clone().all(|digit| *digit == b'0') { + return Ok(0); + } + + let exponent = if exponent_start == bytes.len() { + 0_i64 + } else { + parse_counter_exponent(&bytes[exponent_start + 1..])? + }; + let fraction_len = i64::try_from(fraction_digits.len()) + .map_err(|_| "counter has too many fractional digits")?; + let scale = exponent.saturating_sub(fraction_len); + let total_digits = integer_digits.len() + fraction_digits.len(); + let (kept_digits, appended_zeros) = if scale < 0 { + let removed_digits = usize::try_from(scale.unsigned_abs()) + .map_err(|_| "counter contains a fractional value")?; + if removed_digits > total_digits { + return Err("counter contains a fractional value"); + } + let kept_digits = total_digits - removed_digits; + if integer_digits + .iter() + .chain(fraction_digits) + .skip(kept_digits) + .any(|digit| *digit != b'0') + { + return Err("counter contains a fractional value"); + } + (kept_digits, 0_usize) + } else { + let appended_zeros = usize::try_from(scale).map_err(|_| "counter exceeds the u64 range")?; + if appended_zeros > 20 { + return Err("counter exceeds the u64 range"); + } + (total_digits, appended_zeros) + }; + + let mut counter = 0_u64; + for digit in integer_digits + .iter() + .chain(fraction_digits) + .take(kept_digits) + { + counter = counter + .checked_mul(10) + .and_then(|value| value.checked_add(u64::from(*digit - b'0'))) + .ok_or("counter exceeds the u64 range")?; + } + for _ in 0..appended_zeros { + counter = counter + .checked_mul(10) + .ok_or("counter exceeds the u64 range")?; + } + Ok(counter) +} + +fn parse_counter_exponent(raw: &[u8]) -> std::result::Result { + let (negative, digits) = match raw.first() { + Some(b'+') => (false, &raw[1..]), + Some(b'-') => (true, &raw[1..]), + Some(_) => (false, raw), + None => return Err("counter exponent is missing"), + }; + if digits.is_empty() || !digits.iter().all(u8::is_ascii_digit) { + return Err("counter exponent is invalid"); + } + let exponent = digits.iter().fold(0_i64, |value, digit| { + value + .saturating_mul(10) + .saturating_add(i64::from(*digit - b'0')) + }); + Ok(if negative { + exponent.saturating_neg() + } else { + exponent + }) +} + +#[derive(Debug, Deserialize)] +struct MinioCountSizeWire { + count: WireCounter, + #[serde(rename = "bytes")] + size: WireCounter, +} + +impl From for ReplicationCountSize { + fn from(value: MinioCountSizeWire) -> Self { + Self { + count: value.count.0, + size: value.size.0, + extra: BTreeMap::new(), + } + } +} + +#[derive(Debug, Deserialize)] +struct MinioTimedErrStatsWire { + #[serde(rename = "lastMinute")] + last_minute: MinioCountSizeWire, + #[serde(rename = "lastHour")] + last_hour: MinioCountSizeWire, + totals: MinioCountSizeWire, +} + +impl MinioTimedErrStatsWire { + fn into_totals(self) -> ReplicationCountSize { + let Self { + last_minute, + last_hour, + totals, + } = self; + let _ = (last_minute, last_hour); + totals.into() + } +} + +#[derive(Debug, Deserialize)] +struct MinioQueueMetricWire { + curr: MinioCountSizeWire, + avg: MinioCountSizeWire, + #[serde(default)] + max: WireField, + #[serde(default)] + peak: WireField, +} + +impl MinioQueueMetricWire { + fn try_into_metric(self) -> std::result::Result { + let maximum = match (self.max, self.peak) { + (WireField::Present(max), WireField::Present(peak)) => { + let max: ReplicationCountSize = max.into(); + let peak: ReplicationCountSize = peak.into(); + if max != peak { + return Err("fields `queued.max` and `queued.peak` conflict".into()); + } + max + } + (WireField::Present(max), WireField::Missing) => max.into(), + (WireField::Missing, WireField::Present(peak)) => peak.into(), + (WireField::Missing, WireField::Missing) => { + return Err( + "missing field `queued.peak` or compatibility field `queued.max`".into(), + ); + } + }; + + Ok(ReplicationQueueMetric { + curr: self.curr.into(), + avg: self.avg.into(), + max: maximum, + last_minute: ReplicationCountSize::default(), + extra: BTreeMap::new(), + }) + } +} + +#[derive(Debug, Deserialize)] +struct MinioTargetMetricWire { + #[serde(rename = "replicationCount", default)] + replicated_count: WireField, + #[serde(rename = "completedReplicationSize", default)] + replicated_size: WireField, + #[serde(rename = "limitInBits", default)] + bandwidth_limit: WireField, + #[serde(rename = "currentBandwidth", default)] + current_bandwidth: WireField, + #[serde(default)] + failed: WireField, + #[serde(rename = "pendingReplicationSize", default)] + pending_size: WireField, + #[serde(rename = "replicaSize", default)] + replica_size: WireField, + #[serde(rename = "failedReplicationSize", default)] + failed_size: WireField, + #[serde(rename = "pendingReplicationCount", default)] + pending_count: WireField, + #[serde(rename = "failedReplicationCount", default)] + failed_count: WireField, + #[serde(flatten, default)] + extra: BTreeMap, +} + +impl MinioTargetMetricWire { + fn try_into_metric(self) -> std::result::Result { + let Self { + replicated_count, + replicated_size, + bandwidth_limit, + current_bandwidth, + failed, + pending_size, + replica_size, + failed_size, + pending_count, + failed_count, + extra, + } = self; + let failed = minio_failed_totals(failed); + validate_redundant_failed(&failed, &failed_count, &failed_size)?; + let current_bandwidth = match current_bandwidth { + WireField::Missing => 0.0, + WireField::Present(value) if value.is_finite() && value >= 0.0 => value, + WireField::Present(_) => { + return Err("field `currentBandwidth` must be finite and non-negative".into()); + } + }; + let _ = (pending_size, replica_size, pending_count); + + Ok(ReplicationTargetMetric { + replicated_size: wire_counter_or_zero(replicated_size), + replicated_count: wire_counter_or_zero(replicated_count), + failed, + fail_stats: None, + latency: ReplicationLatencyMetric::default(), + xfer_rate_lrg: ReplicationTransferRate::default(), + xfer_rate_sml: ReplicationTransferRate::default(), + bandwidth_limit_bytes_per_sec: wire_counter_or_zero(bandwidth_limit), + current_bandwidth_bytes_per_sec: current_bandwidth, + latency_scope: Some(ReplicationMetricScope::Unavailable), + bandwidth_scope: None, + extra, + }) + } +} + +#[derive(Debug, Deserialize)] +struct ReplicationMetricsWireEnvelope { + #[serde(rename = "Stats", default)] + minio_stats: WireField>>, + #[serde(rename = "completedReplicationSize", default)] + minio_replicated_size: WireField, + #[serde(rename = "replicaSize", default)] + minio_replica_size: WireField, + #[serde(rename = "replicaCount", default)] + minio_replica_count: WireField, + #[serde(rename = "replicationCount", default)] + minio_replicated_count: WireField, + #[serde(rename = "failed", default)] + minio_failed: WireField, + #[serde(rename = "queued", default)] + minio_queue: WireField, + #[serde(rename = "pendingReplicationSize", default)] + minio_pending_size: WireField, + #[serde(rename = "failedReplicationSize", default)] + minio_failed_size: WireField, + #[serde(rename = "pendingReplicationCount", default)] + minio_pending_count: WireField, + #[serde(rename = "failedReplicationCount", default)] + minio_failed_count: WireField, + #[serde(rename = "stats", default)] + legacy_stats: WireField>, + #[serde(rename = "replica_size", default)] + legacy_replica_size: WireField, + #[serde(rename = "replica_count", default)] + legacy_replica_count: WireField, + #[serde(rename = "replicated_size", default)] + legacy_replicated_size: WireField, + #[serde(rename = "replicated_count", default)] + legacy_replicated_count: WireField, + #[serde(rename = "q_stat", default)] + legacy_queue: WireField, + #[serde(default)] + provider_available: Option, + #[serde(default)] + cluster_complete: Option, + #[serde(default)] + observed_node_count: Option, + #[serde(default)] + expected_node_count: Option, + #[serde(default)] + queue_scope: Option, + #[serde(flatten, default)] + extra: BTreeMap, +} + +impl ReplicationMetricsWireEnvelope { + fn has_minio_fields(&self) -> bool { + self.minio_stats.is_present() + || self.minio_replicated_size.is_present() + || self.minio_replica_size.is_present() + || self.minio_replica_count.is_present() + || self.minio_replicated_count.is_present() + || self.minio_failed.is_present() + || self.minio_queue.is_present() + || self.minio_pending_size.is_present() + || self.minio_failed_size.is_present() + || self.minio_pending_count.is_present() + || self.minio_failed_count.is_present() + } + + fn has_legacy_fields(&self) -> bool { + self.legacy_stats.is_present() + || self.legacy_replica_size.is_present() + || self.legacy_replica_count.is_present() + || self.legacy_replicated_size.is_present() + || self.legacy_replicated_count.is_present() + || self.legacy_queue.is_present() + } + + fn try_into_metrics(self) -> std::result::Result { + match ( + self.minio_stats.is_present(), + self.legacy_stats.is_present(), + ) { + (true, true) => Err("replication metrics response mixes `Stats` and `stats`".into()), + (false, false) => { + Err("missing replication metrics discriminator `Stats` or `stats`".into()) + } + (true, false) => { + if self.has_legacy_fields() { + return Err("MinIO replication metrics contain legacy fields".into()); + } + self.try_into_minio_metrics() + } + (false, true) => { + if self.has_minio_fields() { + return Err("legacy replication metrics contain MinIO fields".into()); + } + self.try_into_legacy_metrics() + } + } + } + + fn try_into_minio_metrics(self) -> std::result::Result { + let stats = self + .minio_stats + .require("Stats")? + .unwrap_or_default() + .into_iter() + .map(|(arn, target)| target.try_into_metric().map(|target| (arn, target))) + .collect::, _>>()?; + let failed = minio_failed_totals(self.minio_failed); + validate_redundant_failed(&failed, &self.minio_failed_count, &self.minio_failed_size)?; + let queue = self.minio_queue.require("queued")?.try_into_metric()?; + let _ = (self.minio_pending_size, self.minio_pending_count, failed); + + Ok(ReplicationMetrics { + stats, + replica_size: wire_counter_or_zero(self.minio_replica_size), + replica_count: wire_counter_or_zero(self.minio_replica_count), + replicated_size: wire_counter_or_zero(self.minio_replicated_size), + replicated_count: wire_counter_or_zero(self.minio_replicated_count), + q_stat: queue, + provider_available: self.provider_available, + cluster_complete: self.cluster_complete, + observed_node_count: self.observed_node_count, + expected_node_count: self.expected_node_count, + queue_scope: self.queue_scope, + extra: self.extra, + }) + } + + fn try_into_legacy_metrics(self) -> std::result::Result { + Ok(ReplicationMetrics { + stats: self.legacy_stats.require("stats")?, + replica_size: self.legacy_replica_size.require("replica_size")?, + replica_count: self.legacy_replica_count.require("replica_count")?, + replicated_size: self.legacy_replicated_size.require("replicated_size")?, + replicated_count: self.legacy_replicated_count.require("replicated_count")?, + q_stat: self.legacy_queue.require("q_stat")?, + provider_available: self.provider_available, + cluster_complete: self.cluster_complete, + observed_node_count: self.observed_node_count, + expected_node_count: self.expected_node_count, + queue_scope: self.queue_scope, + extra: self.extra, + }) + } +} + +fn wire_counter_or_zero(field: WireField) -> u64 { + match field { + WireField::Missing => 0, + WireField::Present(value) => value.0, + } +} + +fn minio_failed_totals(field: WireField) -> ReplicationCountSize { + match field { + WireField::Missing => ReplicationCountSize::default(), + WireField::Present(value) => value.into_totals(), + } +} + +fn validate_redundant_failed( + totals: &ReplicationCountSize, + count: &WireField, + size: &WireField, +) -> std::result::Result<(), String> { + if matches!(count, WireField::Present(value) if value.0 != totals.count) + || matches!(size, WireField::Present(value) if value.0 != totals.size) + { + return Err("replication metrics contain inconsistent failed totals".into()); + } + Ok(()) +} + +impl<'de> Deserialize<'de> for ReplicationMetrics { + fn deserialize(deserializer: D) -> std::result::Result + where + D: serde::Deserializer<'de>, + { + ReplicationMetricsWireEnvelope::deserialize(deserializer)? + .try_into_metrics() + .map_err(serde::de::Error::custom) + } +} + #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct ReplicationMrfTarget { #[serde(rename = "ARN")] @@ -325,4 +809,170 @@ mod tests { let mrf = r#"{"Bucket":"b","Targets":[],"TotalFailedCount":-1,"TotalFailedSize":0,"QueuedCount":0,"QueuedSize":0,"PerObjectEntriesAvailable":false,"RuntimeStatsAvailable":true,"ClusterComplete":false,"ObservedNodeCount":1,"ExpectedNodeCount":2,"DurableBacklogAvailable":false,"DurableCount":0,"DurableSize":0,"PerTargetDurableEntriesAvailable":false}"#; assert!(serde_json::from_str::(mrf).is_err()); } + + #[test] + fn metrics_decode_captured_minio_v1_wire_response() { + let metrics: ReplicationMetrics = serde_json::from_str(include_str!( + "../../tests/fixtures/replication_metrics_minio_v1.json" + )) + .expect("captured MinIO-compatible metrics"); + + let target = metrics + .stats + .get("arn:minio:replication:us-east-1:00000000-0000-0000-0000-000000000000:destination") + .expect("captured target"); + assert_eq!(metrics.replicated_count, 1); + assert_eq!(metrics.replicated_size, 20); + assert_eq!(target.replicated_count, 1); + assert_eq!(target.replicated_size, 20); + assert_eq!( + target.latency_scope, + Some(ReplicationMetricScope::Unavailable) + ); + assert_eq!(metrics.provider_available, Some(true)); + assert_eq!(metrics.cluster_complete, Some(true)); + assert_eq!(metrics.observed_node_count, Some(1)); + assert_eq!(metrics.expected_node_count, Some(1)); + } + + #[test] + fn metrics_use_timed_totals_and_preserve_minio_extensions() { + let metrics: ReplicationMetrics = serde_json::from_str( + r#"{ + "Stats":{"arn:target":{ + "replicationCount":2,"completedReplicationSize":30, + "limitInBits":8000,"currentBandwidth":125.5, + "failed":{ + "lastMinute":{"count":1.0,"bytes":10.0}, + "lastHour":{"count":4.0,"bytes":40.0}, + "totals":{"count":9.0,"bytes":90.0}}, + "failedReplicationCount":9,"failedReplicationSize":90, + "TargetFuture":{"token":"value"}}}, + "completedReplicationSize":30,"replicaSize":4, + "replicaCount":3,"replicationCount":2, + "failed":{ + "lastMinute":{"count":1.0,"bytes":10.0}, + "lastHour":{"count":4.0,"bytes":40.0}, + "totals":{"count":9.0,"bytes":90.0}}, + "queued":{ + "curr":{"count":1.0,"bytes":10.0}, + "avg":{"count":2.0,"bytes":20.0}, + "peak":{"count":3.0,"bytes":30.0}}, + "TopFuture":{"revision":7}} + "#, + ) + .expect("MinIO metrics"); + + let target = &metrics.stats["arn:target"]; + assert_eq!(target.failed.count, 9); + assert_eq!(target.failed.size, 90); + assert_eq!(target.bandwidth_limit_bytes_per_sec, 8000); + assert_eq!(target.current_bandwidth_bytes_per_sec, 125.5); + assert_eq!(target.extra["TargetFuture"]["token"], "value"); + assert_eq!(metrics.q_stat.max.count, 3); + assert_eq!(metrics.extra["TopFuture"]["revision"], 7); + } + + #[test] + fn metrics_accept_queue_peak_or_max_but_reject_conflicts() { + fn payload(queue_tail: &str) -> String { + format!( + r#"{{"Stats":null,"queued":{{ + "curr":{{"count":0,"bytes":0}}, + "avg":{{"count":0,"bytes":0}}, + {queue_tail}}}}}"# + ) + } + + for tail in [ + r#""peak":{"count":2,"bytes":20}"#, + r#""max":{"count":2,"bytes":20}"#, + r#""max":{"count":2,"bytes":20},"peak":{"count":2,"bytes":20}"#, + ] { + let metrics: ReplicationMetrics = + serde_json::from_str(&payload(tail)).expect("compatible queue peak"); + assert_eq!(metrics.q_stat.max.count, 2); + assert_eq!(metrics.q_stat.max.size, 20); + } + + assert!( + serde_json::from_str::(&payload( + r#""max":{"count":2,"bytes":20},"peak":{"count":3,"bytes":20}"#, + )) + .is_err() + ); + } + + #[test] + fn metrics_accept_omitempty_and_lossless_numeric_encodings() { + for (encoded, expected) in [ + ("1e3", 1_000_u64), + ("1.5e1", 15), + ("1.2300e2", 123), + ("0e-400", 0), + ("1000000000000000.0", 1_000_000_000_000_000), + ("9007199254740991.0", 9_007_199_254_740_991), + ("9007199254740992.0", 9_007_199_254_740_992), + ("9223372036854775808.0", 9_223_372_036_854_775_808), + ("1e19", 10_000_000_000_000_000_000), + ("18446744073709551615", u64::MAX), + ("18446744073709551615.0", u64::MAX), + ] { + let payload = format!( + r#"{{"Stats":{{"arn:target":{{"replicationCount":1e3}}}}, + "queued":{{"curr":{{"count":0.0,"bytes":0.0}}, + "avg":{{"count":0,"bytes":0}}, + "peak":{{"count":{encoded},"bytes":42.0}}}}}}"# + ); + let metrics: ReplicationMetrics = + serde_json::from_str(&payload).expect("lossless counter encoding"); + assert_eq!(metrics.replicated_count, 0); + assert_eq!(metrics.stats["arn:target"].replicated_count, 1000); + assert_eq!(metrics.stats["arn:target"].failed.count, 0); + assert_eq!(metrics.q_stat.max.count, expected, "encoded: {encoded}"); + } + + for invalid in [ + "-1", + "-0.0", + "1.5", + "999999999999999.01", + "1e-400", + "18446744073709551616.0", + "1e400", + "NaN", + "Infinity", + "\"1\"", + "null", + ] { + let payload = format!( + r#"{{"Stats":null,"queued":{{ + "curr":{{"count":{invalid},"bytes":0}}, + "avg":{{"count":0,"bytes":0}}, + "peak":{{"count":0,"bytes":0}}}}}}"# + ); + assert!( + serde_json::from_str::(&payload).is_err(), + "accepted invalid counter {invalid}" + ); + } + } + + #[test] + fn metrics_reject_missing_core_mixed_and_inconsistent_minio_fields() { + for payload in [ + r#"{"queued":{"curr":{"count":0,"bytes":0},"avg":{"count":0,"bytes":0},"peak":{"count":0,"bytes":0}}}"#, + r#"{"Stats":null}"#, + r#"{"Stats":null,"stats":{},"queued":{"curr":{"count":0,"bytes":0},"avg":{"count":0,"bytes":0},"peak":{"count":0,"bytes":0}}}"#, + r#"{"Stats":null,"replica_size":0,"queued":{"curr":{"count":0,"bytes":0},"avg":{"count":0,"bytes":0},"peak":{"count":0,"bytes":0}}}"#, + r#"{"Stats":null,"queued":{"avg":{"count":0,"bytes":0},"peak":{"count":0,"bytes":0}}}"#, + r#"{"Stats":null,"queued":{"curr":{"count":0,"bytes":0},"avg":{"count":0,"bytes":0},"peak":{"count":0,"bytes":0}},"failed":{"lastMinute":{"count":0,"bytes":0},"lastHour":{"count":0,"bytes":0}}}"#, + r#"{"Stats":{"arn:target":{"failed":{"lastMinute":{"count":0,"bytes":0},"lastHour":{"count":0,"bytes":0},"totals":{"count":2,"bytes":20}},"failedReplicationCount":1,"failedReplicationSize":20}},"queued":{"curr":{"count":0,"bytes":0},"avg":{"count":0,"bytes":0},"peak":{"count":0,"bytes":0}}}"#, + ] { + assert!( + serde_json::from_str::(payload).is_err(), + "accepted malformed metrics: {payload}" + ); + } + } } diff --git a/crates/core/tests/fixtures/replication_metrics_minio_v1.json b/crates/core/tests/fixtures/replication_metrics_minio_v1.json new file mode 100644 index 00000000..e04a2793 --- /dev/null +++ b/crates/core/tests/fixtures/replication_metrics_minio_v1.json @@ -0,0 +1,36 @@ +{ + "Stats": { + "arn:minio:replication:us-east-1:00000000-0000-0000-0000-000000000000:destination": { + "replicationCount": 1, + "completedReplicationSize": 20, + "limitInBits": 0, + "currentBandwidth": 0.0, + "failed": { + "lastMinute": { "count": 0.0, "bytes": 0 }, + "lastHour": { "count": 0.0, "bytes": 0 }, + "totals": { "count": 0.0, "bytes": 0 } + }, + "failedReplicationSize": 0, + "failedReplicationCount": 0 + } + }, + "completedReplicationSize": 20, + "replicaSize": 0, + "replicaCount": 0, + "replicationCount": 1, + "failed": { + "lastMinute": { "count": 0.0, "bytes": 0 }, + "lastHour": { "count": 0.0, "bytes": 0 }, + "totals": { "count": 0.0, "bytes": 0 } + }, + "queued": { + "curr": { "count": 0.0, "bytes": 0.0 }, + "avg": { "count": 0.0, "bytes": 0.0 }, + "max": { "count": 0.0, "bytes": 0.0 }, + "peak": { "count": 0.0, "bytes": 0.0 } + }, + "provider_available": true, + "cluster_complete": true, + "observed_node_count": 1, + "expected_node_count": 1 +} diff --git a/crates/s3/src/admin.rs b/crates/s3/src/admin.rs index 14904bd9..ed4f8679 100644 --- a/crates/s3/src/admin.rs +++ b/crates/s3/src/admin.rs @@ -1232,6 +1232,64 @@ impl AdminClient { } } + fn unique_redacted_extension_key( + &self, + mut key: String, + next_suffix: &mut BTreeMap, + occupied: impl Fn(&str) -> bool, + ) -> String { + self.redact_admin_credentials(&mut key); + let base = key.clone(); + let suffix = next_suffix.entry(base.clone()).or_insert(2); + while occupied(&key) { + key = format!("{base}#{suffix}"); + *suffix += 1; + } + key + } + + fn sanitize_replication_extension_value(&self, value: &mut serde_json::Value) { + match value { + serde_json::Value::String(value) => self.redact_admin_credentials(value), + serde_json::Value::Array(values) => { + for value in values { + self.sanitize_replication_extension_value(value); + } + } + serde_json::Value::Object(values) => { + let mut sanitized = serde_json::Map::with_capacity(values.len()); + let mut next_suffix = BTreeMap::new(); + for (key, mut value) in std::mem::take(values) { + let key = + self.unique_redacted_extension_key(key, &mut next_suffix, |candidate| { + sanitized.contains_key(candidate) + }); + self.sanitize_replication_extension_value(&mut value); + sanitized.insert(key, value); + } + *values = sanitized; + } + serde_json::Value::Null | serde_json::Value::Bool(_) | serde_json::Value::Number(_) => { + } + } + } + + fn sanitize_replication_extensions( + &self, + extensions: &mut BTreeMap, + ) { + let mut sanitized = BTreeMap::new(); + let mut next_suffix = BTreeMap::new(); + for (key, mut value) in std::mem::take(extensions) { + let key = self.unique_redacted_extension_key(key, &mut next_suffix, |candidate| { + sanitized.contains_key(candidate) + }); + self.sanitize_replication_extension_value(&mut value); + sanitized.insert(key, value); + } + *extensions = sanitized; + } + fn redact_admin_credentials(&self, value: &mut String) { let mut credentials = [&self.access_key, &self.secret_key]; credentials.sort_by_key(|credential| std::cmp::Reverse(credential.len())); @@ -1320,17 +1378,13 @@ impl AdminClient { fn sanitize_replication_metrics(&self, metrics: &mut ReplicationMetrics) { self.sanitize_replication_scope(&mut metrics.queue_scope); - for value in metrics.extra.values_mut() { - self.redact_admin_credentials_in_value(value); - } + self.sanitize_replication_extensions(&mut metrics.extra); for (arn, mut target) in std::mem::take(&mut metrics.stats) { let mut redacted_arn = arn; self.redact_admin_credentials(&mut redacted_arn); self.sanitize_replication_scope(&mut target.latency_scope); self.sanitize_replication_scope(&mut target.bandwidth_scope); - for value in target.extra.values_mut() { - self.redact_admin_credentials_in_value(value); - } + self.sanitize_replication_extensions(&mut target.extra); metrics.stats.insert(redacted_arn, target); } } @@ -1342,13 +1396,9 @@ impl AdminClient { if let ReplicationMetricScope::Unknown(value) = &mut target.observation_scope { self.redact_admin_credentials(value); } - for value in target.extra.values_mut() { - self.redact_admin_credentials_in_value(value); - } - } - for value in mrf.extra.values_mut() { - self.redact_admin_credentials_in_value(value); + self.sanitize_replication_extensions(&mut target.extra); } + self.sanitize_replication_extensions(&mut mrf.extra); } fn map_inspect_archive_error(&self, status: StatusCode, body: &str) -> Error { @@ -5830,7 +5880,9 @@ mod tests { (endpoint, receiver, handle) } - fn start_admin_chunked_overflow_server() -> ( + fn start_admin_chunked_overflow_server( + max_bytes: usize, + ) -> ( String, mpsc::Receiver, mpsc::Receiver<()>, @@ -5850,7 +5902,7 @@ mod tests { let header = b"HTTP/1.1 200 OK\r\ntransfer-encoding: chunked\r\ncontent-type: application/json\r\nconnection: close\r\n\r\n"; let chunk = vec![b'x'; 64 * 1024]; - let mut remaining = MAX_REPLICATION_DIFF_RESPONSE_BYTES; + let mut remaining = max_bytes; let mut write_failed = stream.write_all(header).is_err(); while remaining > 0 && !write_failed { let chunk_len = remaining.min(chunk.len()); @@ -10348,7 +10400,8 @@ mod tests { assert!(matches!(declared, Error::General(message) if message.contains("response limit"))); handle.join().expect("server thread"); - let (endpoint, _receiver, completion) = start_admin_chunked_overflow_server(); + let (endpoint, _receiver, completion) = + start_admin_chunked_overflow_server(MAX_REPLICATION_DIFF_RESPONSE_BYTES); let chunked = anonymous_admin_client_for_endpoint(&endpoint) .replication_diff("source", None) .await @@ -10399,6 +10452,55 @@ mod tests { handle.join().expect("server thread"); } + #[tokio::test] + async fn replication_metrics_decodes_captured_minio_wire_and_sanitizes_extensions() { + let mut body: serde_json::Value = serde_json::from_str(include_str!( + "../../core/tests/fixtures/replication_metrics_minio_v1.json" + )) + .expect("captured metrics fixture"); + let arn = + "arn:minio:replication:us-east-1:00000000-0000-0000-0000-000000000000:destination"; + body["Top-access"] = serde_json::json!({"credential": "secret"}); + body["Stats"][arn]["Target-secret"] = + serde_json::json!({"credential": "access", "nested-access": "secret"}); + let (endpoint, receiver, handle) = start_admin_owned_test_server( + "200 OK", + "application/json", + serde_json::to_string(&body).expect("encoded fixture"), + ); + + let metrics = admin_client_for_endpoint(&endpoint) + .replication_metrics("source bucket") + .await + .expect("captured MinIO-compatible metrics"); + + let target = &metrics.stats[arn]; + assert_eq!(metrics.replicated_count, 1); + assert_eq!(metrics.replicated_size, 20); + assert_eq!(target.replicated_count, 1); + assert_eq!(target.replicated_size, 20); + assert_eq!( + target.latency_scope, + Some(ReplicationMetricScope::Unavailable) + ); + assert!(metrics.extra.contains_key("Top-[REDACTED]")); + assert!(target.extra.contains_key("Target-[REDACTED]")); + assert_eq!(metrics.extra["Top-[REDACTED]"]["credential"], "[REDACTED]"); + assert_eq!( + target.extra["Target-[REDACTED]"]["credential"], + "[REDACTED]" + ); + assert_eq!( + target.extra["Target-[REDACTED]"]["nested-[REDACTED]"], + "[REDACTED]" + ); + assert_eq!( + receiver.recv().expect("request").target, + "/rustfs/admin/v3/replicationmetrics?bucket=source%20bucket" + ); + handle.join().expect("server thread"); + } + #[tokio::test] async fn replication_mrf_preserves_partial_and_explicit_availability() { let body = r#"{"Bucket":"source","Targets":[ @@ -10503,6 +10605,17 @@ mod tests { assert!(matches!(error, Error::General(message) if message.contains("response limit"))); handle.join().expect("server thread"); + let (endpoint, _receiver, completion) = + start_admin_chunked_overflow_server(MAX_REPLICATION_INSPECTION_RESPONSE_BYTES); + let error = anonymous_admin_client_for_endpoint(&endpoint) + .replication_metrics("source") + .await + .expect_err("oversized chunked success"); + assert!(matches!(error, Error::General(message) if message.contains("response limit"))); + completion + .recv_timeout(Duration::from_secs(5)) + .expect("chunked overflow server should complete within its socket timeout"); + let (endpoint, _receiver, handle) = start_admin_declared_length_server( "403 Forbidden", MAX_REPLICATION_INSPECTION_RESPONSE_BYTES + 1,