From 1307350c3a499afe45295da6a6e83681c1ba192f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=A9=AC=E7=99=BB=E5=B1=B1?= Date: Sat, 25 Jul 2026 01:21:06 +0800 Subject: [PATCH] feat(replication): add typed metrics and MRF inspection --- crates/cli/src/commands/replicate.rs | 577 +++++++++++++++++- .../fixtures/output_v3/replication/mrf.json | 28 + .../output_v3/replication/status.json | 46 ++ crates/cli/tests/output_schema_v3.rs | 10 + crates/core/src/admin/mod.rs | 9 +- crates/core/src/admin/replication.rs | 207 ++++++- crates/s3/src/admin.rs | 354 ++++++++++- schemas/output_v3.json | 133 +++- 8 files changed, 1326 insertions(+), 38 deletions(-) create mode 100644 crates/cli/tests/fixtures/output_v3/replication/mrf.json create mode 100644 crates/cli/tests/fixtures/output_v3/replication/status.json diff --git a/crates/cli/src/commands/replicate.rs b/crates/cli/src/commands/replicate.rs index 7f0a13f..ca6763e 100644 --- a/crates/cli/src/commands/replicate.rs +++ b/crates/cli/src/commands/replicate.rs @@ -5,7 +5,10 @@ use clap::{Args, Subcommand}; use comfy_table::{Cell, Table}; -use rc_core::admin::{AdminApi, ReplicationDiff, ReplicationDiffApi, ReplicationDiffEntry}; +use rc_core::admin::{ + AdminApi, ReplicationDiff, ReplicationDiffApi, ReplicationDiffEntry, ReplicationInspectionApi, + ReplicationMetricScope, ReplicationMetrics, ReplicationMrf, +}; use rc_core::replication::{ BucketTarget, BucketTargetCredentials, ReplicationConfiguration, ReplicationDestination, ReplicationResyncStartOptions, ReplicationResyncStartResult, ReplicationResyncStatus, @@ -61,6 +64,9 @@ pub enum ReplicateCommands { /// Show replication status/metrics for a bucket Status(BucketArg), + /// Show aggregate replication MRF backlog for a bucket + Mrf(BucketArg), + /// Scan for object versions that have not replicated Diff(DiffArgs), @@ -422,6 +428,113 @@ struct ReplicationDiffScanOutput { resumable: bool, } +#[derive(Debug, Serialize)] +struct ReplicationInspectionOutput { + schema_version: u8, + #[serde(rename = "type")] + output_type: &'static str, + status: &'static str, + data: T, +} + +#[derive(Debug, Serialize)] +struct ReplicationClusterOutput { + state: &'static str, + observed_nodes: Option, + expected_nodes: Option, +} + +#[derive(Debug, Serialize)] +struct ReplicationStatusData { + operation: &'static str, + bucket: String, + availability: &'static str, + cluster: ReplicationClusterOutput, + queue: ReplicationQueueOutput, + totals: ReplicationTotalsOutput, + targets: Vec, + extensions: BTreeMap, +} + +#[derive(Debug, Serialize)] +struct ReplicationQueueOutput { + count: u64, + size_bytes: u64, + scope: String, +} + +#[derive(Debug, Serialize)] +struct ReplicationTotalsOutput { + replica_count: u64, + replica_size_bytes: u64, + replicated_count: u64, + replicated_size_bytes: u64, +} + +#[derive(Debug, Serialize)] +struct ReplicationStatusTargetOutput { + target_arn: String, + replicated_count: u64, + replicated_size_bytes: u64, + failed_count: u64, + failed_size_bytes: u64, + latency: ReplicationLatencyOutput, + bandwidth: ReplicationBandwidthOutput, + extensions: BTreeMap, +} + +#[derive(Debug, Serialize)] +struct ReplicationLatencyOutput { + average_ms: f64, + current_ms: f64, + maximum_ms: f64, + scope: String, +} + +#[derive(Debug, Serialize)] +struct ReplicationBandwidthOutput { + limit_bytes_per_sec: u64, + current_bytes_per_sec: f64, + scope: String, +} + +#[derive(Debug, Serialize)] +struct ReplicationMrfData { + operation: &'static str, + bucket: String, + runtime_stats_available: bool, + cluster: ReplicationClusterOutput, + failed: ReplicationBacklogOutput, + queue: ReplicationBacklogOutput, + durable_backlog: ReplicationDurableBacklogOutput, + per_object_entries_available: bool, + per_target_durable_entries_available: bool, + targets: Vec, + extensions: BTreeMap, +} + +#[derive(Debug, Serialize)] +struct ReplicationBacklogOutput { + count: u64, + size_bytes: u64, +} + +#[derive(Debug, Serialize)] +struct ReplicationDurableBacklogOutput { + available: bool, + count: u64, + size_bytes: u64, +} + +#[derive(Debug, Serialize)] +struct ReplicationMrfTargetOutput { + target_arn: String, + failed_count: u64, + failed_size_bytes: u64, + observation_scope: String, + extensions: BTreeMap, +} + #[derive(Debug, Serialize)] struct ReplicationDiffErrorOutput { schema_version: u8, @@ -467,6 +580,7 @@ pub async fn execute(args: ReplicateArgs, output_config: OutputConfig) -> ExitCo ReplicateCommands::Update(args) => execute_update(args, output_config).await, ReplicateCommands::List(args) => execute_list(args, output_config).await, ReplicateCommands::Status(args) => execute_status(args, output_config).await, + ReplicateCommands::Mrf(args) => execute_mrf(args, output_config).await, ReplicateCommands::Diff(args) => execute_diff(args, output_config).await, ReplicateCommands::Remove(args) => execute_remove(args, output_config).await, ReplicateCommands::Export(args) => execute_export(args, output_config).await, @@ -1150,29 +1264,346 @@ async fn execute_status(args: BucketArg, output_config: OutputConfig) -> ExitCod Err(code) => return code, }; - match admin_client.replication_metrics(&bucket).await { + execute_status_with_api(&bucket, &admin_client, &formatter).await +} + +async fn execute_status_with_api( + bucket: &str, + api: &dyn ReplicationInspectionApi, + formatter: &Formatter, +) -> ExitCode { + match api.replication_metrics(bucket).await { Ok(metrics) => { if formatter.is_json() { - formatter.json(&metrics); + formatter.json(&replication_status_output(bucket, metrics)); } else { - formatter.println(&format!("Replication metrics for '{alias_name}/{bucket}':")); - match serde_json::to_string_pretty(&metrics) { - Ok(pretty) => formatter.println(&pretty), - Err(error) => { - formatter.error(&format!("Failed to format metrics: {error}")); - return ExitCode::GeneralError; - } + for line in replication_status_lines(bucket, &metrics, formatter) { + formatter.println(&line); } } ExitCode::Success } + Err(error) => emit_replication_inspection_error( + &error, + formatter, + "read replication status", + "replication_status", + ), + } +} + +async fn execute_mrf(args: BucketArg, output_config: OutputConfig) -> ExitCode { + let formatter = Formatter::new(output_config); + let (alias_name, bucket) = match parse_bucket_path(&args.path) { + Ok(parts) => parts, Err(error) => { - formatter.error(&format!("Failed to get replication metrics: {error}")); - ExitCode::GeneralError + return formatter.fail_with_suggestion( + ExitCode::UsageError, + &error, + "Use a bucket path in the form alias/bucket.", + ); } + }; + let admin_client = match setup_admin_client(&alias_name, &formatter) { + Ok(client) => client, + Err(code) => return code, + }; + match admin_client.replication_mrf(&bucket).await { + Ok(mrf) => { + if formatter.is_json() { + formatter.json(&replication_mrf_output(mrf)); + } else { + for line in replication_mrf_lines(&mrf, &formatter) { + formatter.println(&line); + } + } + ExitCode::Success + } + Err(error) => emit_replication_inspection_error( + &error, + &formatter, + "read replication MRF", + "replication_mrf", + ), + } +} + +fn scope_output(scope: Option<&ReplicationMetricScope>) -> String { + scope + .map(|scope| scope.as_str().to_string()) + .unwrap_or_else(|| "legacy_unknown".to_string()) +} + +fn cluster_output( + available: Option, + complete: Option, + observed: Option, + expected: Option, +) -> ReplicationClusterOutput { + ReplicationClusterOutput { + state: match (available, complete) { + (Some(false), _) => "unavailable", + (_, Some(true)) => "complete", + (_, Some(false)) => "partial", + (_, None) => "legacy_unknown", + }, + observed_nodes: observed, + expected_nodes: expected, } } +fn replication_status_output( + bucket: &str, + metrics: ReplicationMetrics, +) -> ReplicationInspectionOutput { + let targets = metrics + .stats + .into_iter() + .map(|(target_arn, target)| { + let failed = target.fail_stats.unwrap_or(target.failed); + ReplicationStatusTargetOutput { + target_arn, + replicated_count: target.replicated_count, + replicated_size_bytes: target.replicated_size, + failed_count: failed.count, + failed_size_bytes: failed.size, + latency: ReplicationLatencyOutput { + average_ms: target.latency.avg, + current_ms: target.latency.curr, + maximum_ms: target.latency.max, + scope: scope_output(target.latency_scope.as_ref()), + }, + bandwidth: ReplicationBandwidthOutput { + limit_bytes_per_sec: target.bandwidth_limit_bytes_per_sec, + current_bytes_per_sec: target.current_bandwidth_bytes_per_sec, + scope: scope_output(target.bandwidth_scope.as_ref()), + }, + extensions: target.extra, + } + }) + .collect(); + ReplicationInspectionOutput { + schema_version: 3, + output_type: "replication", + status: "success", + data: ReplicationStatusData { + operation: "status", + bucket: bucket.to_string(), + availability: match metrics.provider_available { + Some(true) => "available", + Some(false) => "unavailable", + None => "legacy_unknown", + }, + cluster: cluster_output( + metrics.provider_available, + metrics.cluster_complete, + metrics.observed_node_count, + metrics.expected_node_count, + ), + queue: ReplicationQueueOutput { + count: metrics.q_stat.curr.count, + size_bytes: metrics.q_stat.curr.size, + scope: scope_output(metrics.queue_scope.as_ref()), + }, + totals: ReplicationTotalsOutput { + replica_count: metrics.replica_count, + replica_size_bytes: metrics.replica_size, + replicated_count: metrics.replicated_count, + replicated_size_bytes: metrics.replicated_size, + }, + targets, + extensions: metrics.extra, + }, + } +} + +fn replication_status_lines( + bucket: &str, + metrics: &ReplicationMetrics, + formatter: &Formatter, +) -> Vec { + let availability = match metrics.provider_available { + Some(true) => "available", + Some(false) => "unavailable", + None => "legacy/unknown", + }; + let cluster = cluster_output( + metrics.provider_available, + metrics.cluster_complete, + metrics.observed_node_count, + metrics.expected_node_count, + ); + let cluster_detail = match (cluster.observed_nodes, cluster.expected_nodes) { + (Some(observed), Some(expected)) => { + format!("{} ({observed}/{expected} nodes)", cluster.state) + } + _ => cluster.state.to_string(), + }; + let mut lines = vec![ + format!( + "Replication status for '{}': provider={availability}, cluster={}", + formatter.sanitize_text(bucket), + cluster_detail + ), + format!( + "Bucket queue: {} objects, {} bytes (scope: {})", + metrics.q_stat.curr.count, + metrics.q_stat.curr.size, + formatter.sanitize_text(&scope_output(metrics.queue_scope.as_ref())) + ), + format!( + "Totals: replicated {} / {} objects, {} / {} bytes", + metrics.replicated_count, + metrics.replica_count, + metrics.replicated_size, + metrics.replica_size + ), + ]; + if metrics.stats.is_empty() { + lines.push("No target observations were supplied by the server.".into()); + } else { + lines.push("TARGET REPLICATED FAILED LATENCY SCOPE BANDWIDTH SCOPE".into()); + for (arn, target) in &metrics.stats { + let failed = target.fail_stats.as_ref().unwrap_or(&target.failed); + lines.push(format!( + "{} {} / {} bytes {} / {} bytes {} {}", + formatter.sanitize_text(arn), + target.replicated_count, + target.replicated_size, + failed.count, + failed.size, + formatter.sanitize_text(&scope_output(target.latency_scope.as_ref())), + formatter.sanitize_text(&scope_output(target.bandwidth_scope.as_ref())), + )); + } + } + lines +} + +fn replication_mrf_output( + mut mrf: ReplicationMrf, +) -> ReplicationInspectionOutput { + mrf.targets.sort_by(|left, right| left.arn.cmp(&right.arn)); + ReplicationInspectionOutput { + schema_version: 3, + output_type: "replication", + status: "success", + data: ReplicationMrfData { + operation: "mrf", + bucket: mrf.bucket, + runtime_stats_available: mrf.runtime_stats_available, + cluster: cluster_output( + Some(mrf.runtime_stats_available), + Some(mrf.cluster_complete), + Some(mrf.observed_node_count), + Some(mrf.expected_node_count), + ), + failed: ReplicationBacklogOutput { + count: mrf.total_failed_count, + size_bytes: mrf.total_failed_size, + }, + queue: ReplicationBacklogOutput { + count: mrf.queued_count, + size_bytes: mrf.queued_size, + }, + durable_backlog: ReplicationDurableBacklogOutput { + available: mrf.durable_backlog_available, + count: mrf.durable_count, + size_bytes: mrf.durable_size, + }, + per_object_entries_available: mrf.per_object_entries_available, + per_target_durable_entries_available: mrf.per_target_durable_entries_available, + targets: mrf + .targets + .into_iter() + .map(|target| ReplicationMrfTargetOutput { + target_arn: target.arn, + failed_count: target.failed_count, + failed_size_bytes: target.failed_size, + observation_scope: target.observation_scope.as_str().to_string(), + extensions: target.extra, + }) + .collect(), + extensions: mrf.extra, + }, + } +} + +fn replication_mrf_lines(mrf: &ReplicationMrf, formatter: &Formatter) -> Vec { + let cluster_state = if !mrf.runtime_stats_available { + "unavailable" + } else if mrf.cluster_complete { + "complete" + } else { + "partial" + }; + let mut lines = vec![ + format!( + "Replication MRF for '{}': cluster={} ({}/{} nodes)", + formatter.sanitize_text(&mrf.bucket), + cluster_state, + mrf.observed_node_count, + mrf.expected_node_count + ), + format!( + "Failed: {} objects, {} bytes; queued: {} objects, {} bytes", + mrf.total_failed_count, mrf.total_failed_size, mrf.queued_count, mrf.queued_size + ), + format!( + "Durable backlog: {} objects, {} bytes (available: {}); per-object entries available: {}", + mrf.durable_count, + mrf.durable_size, + mrf.durable_backlog_available, + mrf.per_object_entries_available + ), + ]; + let mut targets = mrf.targets.iter().collect::>(); + targets.sort_by(|left, right| left.arn.cmp(&right.arn)); + for target in targets { + lines.push(format!( + "{} failed: {} / {} bytes scope: {}", + formatter.sanitize_text(&target.arn), + target.failed_count, + target.failed_size, + formatter.sanitize_text(target.observation_scope.as_str()) + )); + } + lines +} + +fn emit_replication_inspection_error( + error: &Error, + formatter: &Formatter, + operation: &str, + capability: &'static str, +) -> ExitCode { + let code = ExitCode::from_i32(error.exit_code()).unwrap_or(ExitCode::GeneralError); + let message = format!("Failed to {operation}: {error}"); + if formatter.is_json() { + let output = if matches!(error, Error::UnsupportedFeature(_)) { + ReplicationDiffErrorOutput { + schema_version: 3, + output_type: "replication", + status: "error", + error: ReplicationDiffError::Unsupported(ReplicationDiffUnsupportedError { + error_type: "unsupported_feature", + message, + retryable: false, + capability, + server: None, + suggestion: Some("Upgrade RustFS or verify that the route is enabled."), + }), + } + } else { + replication_diff_error_output(error, code, message) + }; + formatter.json_error(&output); + } else { + formatter.error_with_code(code, &message); + } + code +} + // ==================== Remove ==================== async fn execute_remove(args: RemoveArgs, output_config: OutputConfig) -> ExitCode { @@ -3043,4 +3474,126 @@ mod tests { assert_eq!(code, ExitCode::UsageError); } + + #[test] + fn status_output_preserves_partial_truth_without_inventing_target_queue_or_uptime() { + let metrics: ReplicationMetrics = serde_json::from_str( + r#"{"stats":{"arn:b":{"replicated_size":3,"replicated_count":2, + "failed":{"count":1,"size":4},"fail_stats":{"count":1,"size":4}, + "latency":{"avg":1,"curr":2,"max":3}, + "xfer_rate_lrg":{"avg":0,"curr":0,"peak":0}, + "xfer_rate_sml":{"avg":0,"curr":0,"peak":0}, + "bandwidth_limit_bytes_per_sec":10,"current_bandwidth_bytes_per_sec":5, + "latency_scope":"partial_cluster","bandwidth_scope":"node_local"}}, + "replica_size":4,"replica_count":3,"replicated_size":3,"replicated_count":2, + "q_stat":{"curr":{"count":1,"bytes":4},"avg":{"count":1,"bytes":4}, + "max":{"count":1,"bytes":4},"last_minute":{"count":1,"bytes":4}}, + "provider_available":true,"cluster_complete":false, + "observed_node_count":1,"expected_node_count":2,"queue_scope":"partial_cluster"}"#, + ) + .expect("metrics"); + + let value = serde_json::to_value(replication_status_output("source", metrics)) + .expect("status JSON"); + + assert_eq!(value["data"]["availability"], "available"); + assert_eq!(value["data"]["cluster"]["state"], "partial"); + assert_eq!(value["data"]["queue"]["count"], 1); + assert_eq!(value["data"]["targets"][0]["failed_count"], 1); + assert!(value["data"]["targets"][0].get("queue").is_none()); + assert!(value["data"]["targets"][0].get("uptime").is_none()); + assert!(value["data"].get("healthy").is_none()); + } + + #[test] + fn status_output_marks_legacy_availability_unknown() { + let metrics: ReplicationMetrics = serde_json::from_str( + r#"{"stats":{},"replica_size":0,"replica_count":0, + "replicated_size":0,"replicated_count":0, + "q_stat":{"curr":{"count":0,"bytes":0},"avg":{"count":0,"bytes":0}, + "max":{"count":0,"bytes":0},"last_minute":{"count":0,"bytes":0}}}"#, + ) + .expect("legacy metrics"); + let value = serde_json::to_value(replication_status_output("source", metrics)) + .expect("status JSON"); + assert_eq!(value["data"]["availability"], "legacy_unknown"); + assert_eq!(value["data"]["cluster"]["state"], "legacy_unknown"); + assert_eq!(value["data"]["queue"]["scope"], "legacy_unknown"); + } + + #[test] + fn status_output_keeps_unavailable_provider_distinct_from_valid_empty() { + let unavailable: ReplicationMetrics = serde_json::from_str( + r#"{"stats":{},"replica_size":0,"replica_count":0, + "replicated_size":0,"replicated_count":0, + "q_stat":{"curr":{"count":0,"bytes":0},"avg":{"count":0,"bytes":0}, + "max":{"count":0,"bytes":0},"last_minute":{"count":0,"bytes":0}}, + "provider_available":false,"cluster_complete":false, + "observed_node_count":0,"expected_node_count":2,"queue_scope":"unavailable"}"#, + ) + .expect("unavailable metrics"); + let unavailable = + serde_json::to_value(replication_status_output("source", unavailable)).expect("JSON"); + assert_eq!(unavailable["data"]["availability"], "unavailable"); + assert_eq!(unavailable["data"]["cluster"]["state"], "unavailable"); + + let available: ReplicationMetrics = serde_json::from_str( + r#"{"stats":{},"replica_size":0,"replica_count":0, + "replicated_size":0,"replicated_count":0, + "q_stat":{"curr":{"count":0,"bytes":0},"avg":{"count":0,"bytes":0}, + "max":{"count":0,"bytes":0},"last_minute":{"count":0,"bytes":0}}, + "provider_available":true,"cluster_complete":true, + "observed_node_count":2,"expected_node_count":2,"queue_scope":"cluster_aggregated"}"#, + ) + .expect("valid empty metrics"); + let available = + serde_json::to_value(replication_status_output("source", available)).expect("JSON"); + assert_eq!(available["data"]["availability"], "available"); + assert_eq!(available["data"]["cluster"]["state"], "complete"); + } + + #[test] + fn mrf_output_is_sorted_and_does_not_fabricate_object_rows() { + let mrf: ReplicationMrf = serde_json::from_str( + r#"{"Bucket":"source","Targets":[ + {"ARN":"z","FailedCount":1,"FailedSize":2,"ObservationScope":"node_local"}, + {"ARN":"a","FailedCount":2,"FailedSize":3,"ObservationScope":"partial_cluster"}], + "TotalFailedCount":3,"TotalFailedSize":5,"QueuedCount":4,"QueuedSize":6, + "PerObjectEntriesAvailable":false,"RuntimeStatsAvailable":true, + "ClusterComplete":false,"ObservedNodeCount":1,"ExpectedNodeCount":2, + "DurableBacklogAvailable":true,"DurableCount":7,"DurableSize":8, + "PerTargetDurableEntriesAvailable":false}"#, + ) + .expect("MRF"); + let value = + serde_json::to_value(replication_mrf_output(mrf)).expect("deterministic MRF JSON"); + assert_eq!(value["data"]["targets"][0]["target_arn"], "a"); + assert_eq!(value["data"]["per_object_entries_available"], false); + assert!(value["data"].get("entries").is_none()); + assert!(value["data"]["targets"][0].get("queued_count").is_none()); + } + + #[test] + fn inspection_human_output_sanitizes_server_strings() { + let mrf: ReplicationMrf = serde_json::from_str( + r#"{"Bucket":"source\nspoof","Targets":[ + {"ARN":"arn:\tspoof","FailedCount":0,"FailedSize":0,"ObservationScope":"future\rvalue"}], + "TotalFailedCount":0,"TotalFailedSize":0,"QueuedCount":0,"QueuedSize":0, + "PerObjectEntriesAvailable":false,"RuntimeStatsAvailable":true, + "ClusterComplete":true,"ObservedNodeCount":1,"ExpectedNodeCount":1, + "DurableBacklogAvailable":false,"DurableCount":0,"DurableSize":0, + "PerTargetDurableEntriesAvailable":false}"#, + ) + .expect("MRF"); + let formatter = Formatter::new(OutputConfig { + no_color: true, + ..OutputConfig::default() + }); + let output = replication_mrf_lines(&mrf, &formatter).join("\n"); + assert!(output.contains("source\\nspoof")); + assert!(output.contains("arn:\\tspoof")); + assert!(output.contains("future\\rvalue")); + assert!(!output.contains('\r')); + assert!(!output.contains('\t')); + } } diff --git a/crates/cli/tests/fixtures/output_v3/replication/mrf.json b/crates/cli/tests/fixtures/output_v3/replication/mrf.json new file mode 100644 index 0000000..71cd652 --- /dev/null +++ b/crates/cli/tests/fixtures/output_v3/replication/mrf.json @@ -0,0 +1,28 @@ +{ + "schema_version": 3, + "type": "replication", + "status": "success", + "data": { + "operation": "mrf", + "bucket": "source", + "runtime_stats_available": true, + "cluster": { + "state": "partial", + "observed_nodes": 1, + "expected_nodes": 2 + }, + "failed": {"count": 2, "size_bytes": 20}, + "queue": {"count": 1, "size_bytes": 10}, + "durable_backlog": {"available": true, "count": 3, "size_bytes": 30}, + "per_object_entries_available": false, + "per_target_durable_entries_available": false, + "targets": [{ + "target_arn": "arn:rustfs:replication::target", + "failed_count": 2, + "failed_size_bytes": 20, + "observation_scope": "partial_cluster", + "extensions": {} + }], + "extensions": {} + } +} diff --git a/crates/cli/tests/fixtures/output_v3/replication/status.json b/crates/cli/tests/fixtures/output_v3/replication/status.json new file mode 100644 index 0000000..b89b238 --- /dev/null +++ b/crates/cli/tests/fixtures/output_v3/replication/status.json @@ -0,0 +1,46 @@ +{ + "schema_version": 3, + "type": "replication", + "status": "success", + "data": { + "operation": "status", + "bucket": "source", + "availability": "available", + "cluster": { + "state": "partial", + "observed_nodes": 1, + "expected_nodes": 2 + }, + "queue": { + "count": 1, + "size_bytes": 10, + "scope": "partial_cluster" + }, + "totals": { + "replica_count": 4, + "replica_size_bytes": 40, + "replicated_count": 3, + "replicated_size_bytes": 30 + }, + "targets": [{ + "target_arn": "arn:rustfs:replication::target", + "replicated_count": 3, + "replicated_size_bytes": 30, + "failed_count": 1, + "failed_size_bytes": 10, + "latency": { + "average_ms": 1.0, + "current_ms": 2.0, + "maximum_ms": 3.0, + "scope": "partial_cluster" + }, + "bandwidth": { + "limit_bytes_per_sec": 100, + "current_bytes_per_sec": 50.0, + "scope": "node_local" + }, + "extensions": {} + }], + "extensions": {} + } +} diff --git a/crates/cli/tests/output_schema_v3.rs b/crates/cli/tests/output_schema_v3.rs index 0891f92..f543644 100644 --- a/crates/cli/tests/output_schema_v3.rs +++ b/crates/cli/tests/output_schema_v3.rs @@ -147,6 +147,16 @@ fn every_v3_family_has_valid_contract_fixtures() { } } +#[test] +fn replication_status_and_mrf_fixtures_are_valid() { + let validator = load_validator(3); + for case in ["status", "mrf"] { + let path = fixture_path("replication", case); + let value = load_json(&path); + assert_valid(&validator, &value, &path.display().to_string()); + } +} + #[test] fn multipart_partial_fixture_preserves_successes_and_per_upload_errors() { let validator = load_validator(3); diff --git a/crates/core/src/admin/mod.rs b/crates/core/src/admin/mod.rs index ed7e9fc..8d6ab4b 100644 --- a/crates/core/src/admin/mod.rs +++ b/crates/core/src/admin/mod.rs @@ -95,7 +95,11 @@ pub use oidc::{ OidcValidationResult, }; pub use replication::{ - MAX_REPLICATION_DIFF_RESPONSE_BYTES, ReplicationDiff, ReplicationDiffApi, ReplicationDiffEntry, + MAX_REPLICATION_DIFF_RESPONSE_BYTES, MAX_REPLICATION_INSPECTION_RESPONSE_BYTES, + ReplicationCountSize, ReplicationDiff, ReplicationDiffApi, ReplicationDiffEntry, + ReplicationInspectionApi, ReplicationLatencyMetric, ReplicationMetricScope, ReplicationMetrics, + ReplicationMrf, ReplicationMrfTarget, ReplicationQueueMetric, ReplicationTargetMetric, + ReplicationTransferRate, }; pub use site::{ MAX_SITE_REPLICATION_CA_CERT_BYTES, MAX_SITE_REPLICATION_ERROR_RESPONSE_BYTES, @@ -324,9 +328,6 @@ pub trait AdminApi: Send + Sync { /// Remove a remote replication target async fn remove_remote_target(&self, bucket: &str, arn: &str) -> Result<()>; - /// Get replication metrics for a bucket - async fn replication_metrics(&self, bucket: &str) -> Result; - // ==================== Service Control Operations ==================== /// Request a service action (restart, stop, freeze, unfreeze) diff --git a/crates/core/src/admin/replication.rs b/crates/core/src/admin/replication.rs index 44f7912..bc72f2c 100644 --- a/crates/core/src/admin/replication.rs +++ b/crates/core/src/admin/replication.rs @@ -1,4 +1,4 @@ -//! Typed contracts for read-only replication diff inspection. +//! Typed contracts for read-only replication inspection. use std::collections::BTreeMap; @@ -11,6 +11,189 @@ use crate::Result; /// Maximum encoded size accepted for one replication diff response. pub const MAX_REPLICATION_DIFF_RESPONSE_BYTES: usize = 8 * 1024 * 1024; +/// Maximum encoded size accepted for metrics and MRF responses. +pub const MAX_REPLICATION_INSPECTION_RESPONSE_BYTES: usize = 8 * 1024 * 1024; + +/// Scope of a replication observation. Unknown values are retained for forward compatibility. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ReplicationMetricScope { + Unavailable, + NodeLocal, + ClusterAggregated, + PartialCluster, + Unknown(String), +} + +impl ReplicationMetricScope { + pub fn as_str(&self) -> &str { + match self { + Self::Unavailable => "unavailable", + Self::NodeLocal => "node_local", + Self::ClusterAggregated => "cluster_aggregated", + Self::PartialCluster => "partial_cluster", + Self::Unknown(value) => value, + } + } +} + +impl Serialize for ReplicationMetricScope { + fn serialize( + &self, + serializer: S, + ) -> std::result::Result { + serializer.serialize_str(self.as_str()) + } +} + +impl<'de> Deserialize<'de> for ReplicationMetricScope { + fn deserialize>( + deserializer: D, + ) -> std::result::Result { + let value = String::deserialize(deserializer)?; + Ok(match value.as_str() { + "unavailable" => Self::Unavailable, + "node_local" => Self::NodeLocal, + "cluster_aggregated" => Self::ClusterAggregated, + "partial_cluster" => Self::PartialCluster, + _ => Self::Unknown(value), + }) + } +} + +#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)] +pub struct ReplicationCountSize { + pub count: u64, + #[serde(rename = "bytes", alias = "size")] + pub size: u64, + #[serde(flatten, default)] + pub extra: BTreeMap, +} + +#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)] +pub struct ReplicationQueueMetric { + pub curr: ReplicationCountSize, + pub avg: ReplicationCountSize, + pub max: ReplicationCountSize, + pub last_minute: ReplicationCountSize, + #[serde(flatten, default)] + pub extra: BTreeMap, +} + +#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)] +pub struct ReplicationLatencyMetric { + pub avg: f64, + pub curr: f64, + pub max: f64, + #[serde(flatten, default)] + pub extra: BTreeMap, +} + +#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)] +pub struct ReplicationTransferRate { + pub avg: f64, + pub curr: f64, + pub peak: f64, + #[serde(flatten, default)] + pub extra: BTreeMap, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct ReplicationTargetMetric { + pub replicated_size: u64, + pub replicated_count: u64, + pub failed: ReplicationCountSize, + #[serde(default)] + pub fail_stats: Option, + pub latency: ReplicationLatencyMetric, + pub xfer_rate_lrg: ReplicationTransferRate, + pub xfer_rate_sml: ReplicationTransferRate, + pub bandwidth_limit_bytes_per_sec: u64, + pub current_bandwidth_bytes_per_sec: f64, + #[serde(default)] + pub latency_scope: Option, + #[serde(default)] + pub bandwidth_scope: Option, + #[serde(flatten, default)] + pub extra: BTreeMap, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct ReplicationMetrics { + pub stats: BTreeMap, + pub replica_size: u64, + pub replica_count: u64, + pub replicated_size: u64, + pub replicated_count: u64, + pub q_stat: ReplicationQueueMetric, + #[serde(default)] + pub provider_available: Option, + #[serde(default)] + pub cluster_complete: Option, + #[serde(default)] + pub observed_node_count: Option, + #[serde(default)] + pub expected_node_count: Option, + #[serde(default)] + pub queue_scope: Option, + #[serde(flatten, default)] + pub extra: BTreeMap, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct ReplicationMrfTarget { + #[serde(rename = "ARN")] + pub arn: String, + #[serde(rename = "FailedCount")] + pub failed_count: u64, + #[serde(rename = "FailedSize")] + pub failed_size: u64, + #[serde(rename = "ObservationScope")] + pub observation_scope: ReplicationMetricScope, + #[serde(flatten, default)] + pub extra: BTreeMap, +} + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +pub struct ReplicationMrf { + #[serde(rename = "Bucket")] + pub bucket: String, + #[serde(rename = "Targets")] + pub targets: Vec, + #[serde(rename = "TotalFailedCount")] + pub total_failed_count: u64, + #[serde(rename = "TotalFailedSize")] + pub total_failed_size: u64, + #[serde(rename = "QueuedCount")] + pub queued_count: u64, + #[serde(rename = "QueuedSize")] + pub queued_size: u64, + #[serde(rename = "PerObjectEntriesAvailable")] + pub per_object_entries_available: bool, + #[serde(rename = "RuntimeStatsAvailable")] + pub runtime_stats_available: bool, + #[serde(rename = "ClusterComplete")] + pub cluster_complete: bool, + #[serde(rename = "ObservedNodeCount")] + pub observed_node_count: u32, + #[serde(rename = "ExpectedNodeCount")] + pub expected_node_count: u32, + #[serde(rename = "DurableBacklogAvailable")] + pub durable_backlog_available: bool, + #[serde(rename = "DurableCount")] + pub durable_count: u64, + #[serde(rename = "DurableSize")] + pub durable_size: u64, + #[serde(rename = "PerTargetDurableEntriesAvailable")] + pub per_target_durable_entries_available: bool, + #[serde(flatten, default)] + pub extra: BTreeMap, +} + +#[async_trait] +pub trait ReplicationInspectionApi: Send + Sync { + async fn replication_metrics(&self, bucket: &str) -> Result; + async fn replication_mrf(&self, bucket: &str) -> Result; +} /// A bounded, on-demand scan of object versions that have not replicated. #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] @@ -120,4 +303,26 @@ mod tests { let payload = r#"{"Entries":[]}"#; assert!(serde_json::from_str::(payload).is_err()); } + + #[test] + fn metrics_distinguish_legacy_metadata_and_preserve_unknown_scope() { + let legacy: ReplicationMetrics = serde_json::from_str(r#"{"stats":{},"replica_size":0,"replica_count":0,"replicated_size":0,"replicated_count":0,"q_stat":{"curr":{"count":0,"size":0},"avg":{"count":0,"size":0},"max":{"count":0,"size":0},"last_minute":{"count":0,"size":0}}}"#).expect("legacy metrics"); + assert_eq!(legacy.provider_available, None); + assert_eq!(legacy.cluster_complete, None); + + let current: ReplicationMetrics = serde_json::from_str(r#"{"stats":{},"replica_size":0,"replica_count":0,"replicated_size":0,"replicated_count":0,"q_stat":{"curr":{"count":0,"size":0},"avg":{"count":0,"size":0},"max":{"count":0,"size":0},"last_minute":{"count":0,"size":0}},"provider_available":true,"cluster_complete":false,"observed_node_count":1,"expected_node_count":2,"queue_scope":"future_scope"}"#).expect("current metrics"); + assert_eq!(current.provider_available, Some(true)); + assert_eq!( + current.queue_scope, + Some(ReplicationMetricScope::Unknown("future_scope".into())) + ); + } + + #[test] + fn metrics_and_mrf_reject_negative_counters() { + let metrics = r#"{"stats":{},"replica_size":-1,"replica_count":0,"replicated_size":0,"replicated_count":0,"q_stat":{"curr":{"count":0,"size":0},"avg":{"count":0,"size":0},"max":{"count":0,"size":0},"last_minute":{"count":0,"size":0}}}"#; + assert!(serde_json::from_str::(metrics).is_err()); + 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()); + } } diff --git a/crates/s3/src/admin.rs b/crates/s3/src/admin.rs index c6db73e..ee9d554 100644 --- a/crates/s3/src/admin.rs +++ b/crates/s3/src/admin.rs @@ -34,15 +34,16 @@ use rc_core::admin::{ MAX_IAM_POLICY_DETACH_REQUEST_BYTES, MAX_IAM_POLICY_DETACH_RESPONSE_BYTES, MAX_IAM_POLICY_ENTITIES_RESPONSE_BYTES, MAX_METRICS_LINE_BYTES, MAX_METRICS_RESPONSE_BYTES, MAX_METRICS_SAMPLES, MAX_OIDC_RESPONSE_BYTES, MAX_REPLICATION_DIFF_RESPONSE_BYTES, - MAX_SITE_REPLICATION_ERROR_RESPONSE_BYTES, MAX_SITE_REPLICATION_REQUEST_BYTES, - MAX_SITE_REPLICATION_SUCCESS_RESPONSE_BYTES, ManualTransitionRunRequest, - ManualTransitionRunResponse, MetricsBatch, MetricsQuery, ModuleSwitches, ObservabilityApi, - OidcMutationApi, OidcMutationRequest, OidcMutationResult, OidcProvider, OidcProviderList, - OidcReadApi, OidcValidationRequest, OidcValidationResult, PeerSiteSpec, Policy, - PolicyDetachEntity, PolicyDetachRequest, PolicyDetachResult, PolicyEntitiesQuery, - PolicyEntitiesResult, PolicyEntity, PolicyInfo, PoolStatus, PoolTarget, RealtimeMetrics, - RebalanceStartResult, RebalanceStatus, ReplicateEditStatus, ReplicationDiff, - ReplicationDiffApi, RuntimeCapabilitiesSnapshot, RuntimeCapabilityStatus, ScannerStatus, + MAX_REPLICATION_INSPECTION_RESPONSE_BYTES, MAX_SITE_REPLICATION_ERROR_RESPONSE_BYTES, + MAX_SITE_REPLICATION_REQUEST_BYTES, MAX_SITE_REPLICATION_SUCCESS_RESPONSE_BYTES, + ManualTransitionRunRequest, ManualTransitionRunResponse, MetricsBatch, MetricsQuery, + ModuleSwitches, ObservabilityApi, OidcMutationApi, OidcMutationRequest, OidcMutationResult, + OidcProvider, OidcProviderList, OidcReadApi, OidcValidationRequest, OidcValidationResult, + PeerSiteSpec, Policy, PolicyDetachEntity, PolicyDetachRequest, PolicyDetachResult, + PolicyEntitiesQuery, PolicyEntitiesResult, PolicyEntity, PolicyInfo, PoolStatus, PoolTarget, + RealtimeMetrics, RebalanceStartResult, RebalanceStatus, ReplicateEditStatus, ReplicationDiff, + ReplicationDiffApi, ReplicationInspectionApi, ReplicationMetricScope, ReplicationMetrics, + ReplicationMrf, RuntimeCapabilitiesSnapshot, RuntimeCapabilityStatus, ScannerStatus, ServiceAccount, ServiceAccountCreateResponse, ServiceActionResult, SiteRemoveSpec, SiteReplicationInfo, SiteReplicationPeer, SiteReplicationResyncOperation, SiteReplicationResyncStatus, SiteStatusOptions, StorageInfo, UpdateGroupMembersRequest, @@ -1179,6 +1180,89 @@ impl AdminClient { Error::UnsupportedFeature(reason) } + fn map_replication_inspection_error(&self, status: StatusCode, body: &str) -> Error { + if matches!(status, StatusCode::FORBIDDEN | StatusCode::UNAUTHORIZED) { + return Error::Auth("Authentication failed for replication inspection".into()); + } + if matches!( + status, + StatusCode::METHOD_NOT_ALLOWED | StatusCode::NOT_IMPLEMENTED + ) { + return Error::UnsupportedFeature( + "Replication inspection is not supported by this server".into(), + ); + } + if status != StatusCode::NOT_FOUND { + let mut redacted = body.to_string(); + self.redact_admin_credentials(&mut redacted); + return self.map_error(status, &redacted); + } + let structured = parse_admin_error(body); + let mut safe_message = structured + .as_ref() + .and_then(|error| error.message.clone()) + .filter(|message| !message.trim().is_empty()); + if let Some(message) = &mut safe_message { + self.redact_admin_credentials(message); + } + if structured.as_ref().is_some_and(|error| { + matches!( + error.code.as_deref(), + Some( + "NoSuchBucket" + | "ReplicationConfigurationNotFoundError" + | "ReplicationConfigurationNotFound" + ) + ) + }) { + return Error::NotFound(safe_message.unwrap_or_else(|| { + "Bucket or replication configuration was not found".to_string() + })); + } + Error::UnsupportedFeature( + safe_message.unwrap_or_else(|| "Replication inspection route was not found".into()), + ) + } + + fn sanitize_replication_scope(&self, scope: &mut Option) { + if let Some(ReplicationMetricScope::Unknown(value)) = scope { + self.redact_admin_credentials(value); + } + } + + 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); + } + 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); + } + metrics.stats.insert(redacted_arn, target); + } + } + + fn sanitize_replication_mrf(&self, mrf: &mut ReplicationMrf) { + self.redact_admin_credentials(&mut mrf.bucket); + for target in &mut mrf.targets { + self.redact_admin_credentials(&mut target.arn); + 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); + } + } + fn map_iam_policy_entities_error(&self, status: StatusCode, _body: &str) -> Error { match status { StatusCode::NOT_FOUND @@ -4521,12 +4605,6 @@ impl AdminApi for AdminClient { .await } - async fn replication_metrics(&self, bucket: &str) -> Result { - let query: &[(&str, &str)] = &[("bucket", bucket)]; - self.request(Method::GET, "/replicationmetrics", Some(query), None) - .await - } - async fn service_action(&self, action: &str) -> Result { let query: &[(&str, &str)] = &[("action", action)]; self.request(Method::POST, "/service", Some(query), None) @@ -4688,6 +4766,96 @@ impl ReplicationDiffApi for AdminClient { } } +#[async_trait] +impl ReplicationInspectionApi for AdminClient { + async fn replication_metrics(&self, bucket: &str) -> Result { + let query = [("bucket", bucket)]; + let mut metrics: ReplicationMetrics = self + .request_bounded_json( + Method::GET, + "/replicationmetrics", + Some(&query), + None, + BoundedJsonResponse { + max_bytes: MAX_REPLICATION_INSPECTION_RESPONSE_BYTES, + name: "Replication metrics response", + error_mapper: Self::map_replication_inspection_error, + }, + ) + .await?; + for target in metrics.stats.values() { + if target + .fail_stats + .as_ref() + .is_some_and(|authoritative| authoritative != &target.failed) + { + return Err(Error::General( + "Replication metrics response contains inconsistent failed totals".into(), + )); + } + if [ + target.latency.avg, + target.latency.curr, + target.latency.max, + target.xfer_rate_lrg.avg, + target.xfer_rate_lrg.curr, + target.xfer_rate_lrg.peak, + target.xfer_rate_sml.avg, + target.xfer_rate_sml.curr, + target.xfer_rate_sml.peak, + target.current_bandwidth_bytes_per_sec, + ] + .into_iter() + .any(|value| !value.is_finite() || value < 0.0) + { + return Err(Error::General( + "Replication metrics response contains invalid negative observations".into(), + )); + } + } + self.sanitize_replication_metrics(&mut metrics); + Ok(metrics) + } + + async fn replication_mrf(&self, bucket: &str) -> Result { + let query = [("bucket", bucket)]; + let mut mrf: ReplicationMrf = self + .request_bounded_json( + Method::GET, + "/replication/mrf", + Some(&query), + None, + BoundedJsonResponse { + max_bytes: MAX_REPLICATION_INSPECTION_RESPONSE_BYTES, + name: "Replication MRF response", + error_mapper: Self::map_replication_inspection_error, + }, + ) + .await?; + if mrf.bucket != bucket { + return Err(Error::General( + "Replication MRF response bucket does not match the request".into(), + )); + } + let target_count = mrf.targets.iter().try_fold(0_u64, |total, target| { + total.checked_add(target.failed_count) + }); + let target_size = mrf + .targets + .iter() + .try_fold(0_u64, |total, target| total.checked_add(target.failed_size)); + if target_count != Some(mrf.total_failed_count) + || target_size != Some(mrf.total_failed_size) + { + return Err(Error::General( + "Replication MRF response contains inconsistent target totals".into(), + )); + } + self.sanitize_replication_mrf(&mut mrf); + Ok(mrf) + } +} + #[cfg(test)] mod tests { use super::*; @@ -9166,6 +9334,162 @@ mod tests { .expect("chunked overflow server should complete within its socket timeout"); } + #[tokio::test] + async fn replication_metrics_reads_typed_truth_and_redacts_credentials() { + let body = r#"{ + "stats":{"arn:secret":{"replicated_size":30,"replicated_count":3, + "failed":{"count":2,"size":20},"fail_stats":{"count":2,"size":20}, + "latency":{"avg":1.0,"curr":2.0,"max":3.0}, + "xfer_rate_lrg":{"avg":4.0,"curr":5.0,"peak":6.0}, + "xfer_rate_sml":{"avg":7.0,"curr":8.0,"peak":9.0}, + "bandwidth_limit_bytes_per_sec":100, + "current_bandwidth_bytes_per_sec":50.0, + "latency_scope":"partial_cluster","bandwidth_scope":"node_local", + "Future":"access"}}, + "replica_size":40,"replica_count":4,"replicated_size":30,"replicated_count":3, + "q_stat":{"curr":{"count":1,"bytes":10},"avg":{"count":1,"bytes":10}, + "max":{"count":2,"bytes":20},"last_minute":{"count":1,"bytes":10}}, + "provider_available":true,"cluster_complete":false, + "observed_node_count":1,"expected_node_count":2,"queue_scope":"partial_cluster" + }"#; + let (endpoint, receiver, handle) = start_admin_test_server("200 OK", body); + let client = admin_client_for_endpoint(&endpoint); + + let metrics = client + .replication_metrics("source bucket") + .await + .expect("typed metrics"); + + assert_eq!(metrics.provider_available, Some(true)); + assert_eq!(metrics.q_stat.curr.size, 10); + assert!(metrics.stats.contains_key("arn:[REDACTED]")); + assert_eq!( + metrics.stats["arn:[REDACTED]"].extra["Future"], + "[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":[ + {"ARN":"arn:a","FailedCount":2,"FailedSize":20,"ObservationScope":"partial_cluster"} + ],"TotalFailedCount":2,"TotalFailedSize":20,"QueuedCount":1,"QueuedSize":10, + "PerObjectEntriesAvailable":false,"RuntimeStatsAvailable":true, + "ClusterComplete":false,"ObservedNodeCount":1,"ExpectedNodeCount":2, + "DurableBacklogAvailable":true,"DurableCount":3,"DurableSize":30, + "PerTargetDurableEntriesAvailable":false}"#; + let (endpoint, receiver, handle) = start_admin_test_server("200 OK", body); + + let mrf = anonymous_admin_client_for_endpoint(&endpoint) + .replication_mrf("source") + .await + .expect("typed MRF"); + + assert!(!mrf.cluster_complete); + assert!(!mrf.per_object_entries_available); + assert_eq!(mrf.queued_count, 1); + assert_eq!( + receiver.recv().expect("request").target, + "/rustfs/admin/v3/replication/mrf?bucket=source" + ); + handle.join().expect("server thread"); + } + + #[tokio::test] + async fn replication_inspection_distinguishes_errors_and_rejects_bad_truth() { + for (status, body, expected) in [ + ( + "403 Forbidden", + r#"{"Code":"AccessDenied","Message":"denied"}"#, + "auth", + ), + ( + "404 Not Found", + r#"{"Code":"ReplicationConfigurationNotFoundError","Message":"absent"}"#, + "not_found", + ), + ( + "404 Not Found", + r#"{"message":"route absent"}"#, + "unsupported", + ), + ] { + let (endpoint, _receiver, handle) = start_admin_test_server(status, body); + let error = anonymous_admin_client_for_endpoint(&endpoint) + .replication_metrics("source") + .await + .expect_err("inspection error"); + match expected { + "auth" => assert!(matches!(error, Error::Auth(_))), + "not_found" => assert!(matches!(error, Error::NotFound(_))), + "unsupported" => assert!(matches!(error, Error::UnsupportedFeature(_))), + _ => unreachable!(), + } + handle.join().expect("server thread"); + } + + let (endpoint, _receiver, handle) = + start_admin_test_server("404 Not Found", r#"{"message":"route access absent"}"#); + let redacted = admin_client_for_endpoint(&endpoint) + .replication_metrics("source") + .await + .expect_err("unsupported response"); + assert!(!redacted.to_string().contains("access")); + assert!(redacted.to_string().contains("[REDACTED]")); + handle.join().expect("server thread"); + + let (endpoint, _receiver, handle) = start_admin_test_server("200 OK", "not-json"); + assert!(matches!( + anonymous_admin_client_for_endpoint(&endpoint) + .replication_metrics("source") + .await + .expect_err("malformed response"), + Error::Json(_) + )); + handle.join().expect("server thread"); + + let inconsistent = r#"{"Bucket":"source","Targets":[{"ARN":"a","FailedCount":1,"FailedSize":1,"ObservationScope":"node_local"}],"TotalFailedCount":2,"TotalFailedSize":1,"QueuedCount":0,"QueuedSize":0,"PerObjectEntriesAvailable":false,"RuntimeStatsAvailable":true,"ClusterComplete":true,"ObservedNodeCount":1,"ExpectedNodeCount":1,"DurableBacklogAvailable":false,"DurableCount":0,"DurableSize":0,"PerTargetDurableEntriesAvailable":false}"#; + let (endpoint, _receiver, handle) = start_admin_test_server("200 OK", inconsistent); + assert!(matches!( + anonymous_admin_client_for_endpoint(&endpoint) + .replication_mrf("source") + .await + .expect_err("inconsistent totals"), + Error::General(message) if message.contains("inconsistent") + )); + handle.join().expect("server thread"); + } + + #[tokio::test] + async fn replication_inspection_bounds_error_and_success_bodies() { + let (endpoint, _receiver, handle) = start_admin_declared_length_server( + "200 OK", + MAX_REPLICATION_INSPECTION_RESPONSE_BYTES + 1, + ); + let error = anonymous_admin_client_for_endpoint(&endpoint) + .replication_metrics("source") + .await + .expect_err("oversized success"); + assert!(matches!(error, Error::General(message) if message.contains("response limit"))); + handle.join().expect("server thread"); + + let (endpoint, _receiver, handle) = start_admin_declared_length_server( + "403 Forbidden", + MAX_REPLICATION_INSPECTION_RESPONSE_BYTES + 1, + ); + let error = anonymous_admin_client_for_endpoint(&endpoint) + .replication_mrf("source") + .await + .expect_err("oversized error"); + assert!(matches!(error, Error::General(message) if message.contains("response limit"))); + handle.join().expect("server thread"); + } + #[tokio::test] async fn observability_routes_distinguish_permission_denial_from_unsupported() { for (status, expected_auth) in [("403 Forbidden", true), ("404 Not Found", false)] { diff --git a/schemas/output_v3.json b/schemas/output_v3.json index 8a5a864..9a7ad2f 100644 --- a/schemas/output_v3.json +++ b/schemas/output_v3.json @@ -1255,19 +1255,140 @@ } }, "replicationData": { + "oneOf": [ + { "$ref": "#/definitions/replicationDiffData" }, + { "$ref": "#/definitions/replicationStatusData" }, + { "$ref": "#/definitions/replicationMrfData" } + ] + }, + "replicationDiffData": { "type": "object", - "required": [ - "operation", "bucket", "prefix", "entries", "scan", "extensions" - ], + "required": ["operation", "bucket", "prefix", "entries", "scan", "extensions"], "properties": { "operation": { "const": "diff" }, "bucket": { "type": "string", "minLength": 1 }, "prefix": { "$ref": "#/definitions/nullableString" }, - "entries": { + "entries": { "type": "array", "items": { "$ref": "#/definitions/replicationDiffEntry" } }, + "scan": { "$ref": "#/definitions/replicationDiffScan" }, + "extensions": { "type": "object" } + } + }, + "replicationCluster": { + "type": "object", + "required": ["state", "observed_nodes", "expected_nodes"], + "properties": { + "state": { "enum": ["complete", "partial", "unavailable", "legacy_unknown"] }, + "observed_nodes": { "type": ["integer", "null"], "minimum": 0 }, + "expected_nodes": { "type": ["integer", "null"], "minimum": 0 } + } + }, + "replicationBacklog": { + "type": "object", + "required": ["count", "size_bytes"], + "properties": { + "count": { "type": "integer", "minimum": 0 }, + "size_bytes": { "type": "integer", "minimum": 0 } + } + }, + "replicationStatusData": { + "type": "object", + "required": ["operation", "bucket", "availability", "cluster", "queue", "totals", "targets", "extensions"], + "properties": { + "operation": { "const": "status" }, + "bucket": { "type": "string", "minLength": 1 }, + "availability": { "enum": ["available", "unavailable", "legacy_unknown"] }, + "cluster": { "$ref": "#/definitions/replicationCluster" }, + "queue": { + "type": "object", + "required": ["count", "size_bytes", "scope"], + "properties": { + "count": { "type": "integer", "minimum": 0 }, + "size_bytes": { "type": "integer", "minimum": 0 }, + "scope": { "type": "string", "minLength": 1 } + } + }, + "totals": { + "type": "object", + "required": ["replica_count", "replica_size_bytes", "replicated_count", "replicated_size_bytes"], + "properties": { + "replica_count": { "type": "integer", "minimum": 0 }, + "replica_size_bytes": { "type": "integer", "minimum": 0 }, + "replicated_count": { "type": "integer", "minimum": 0 }, + "replicated_size_bytes": { "type": "integer", "minimum": 0 } + } + }, + "targets": { "type": "array", - "items": { "$ref": "#/definitions/replicationDiffEntry" } + "items": { + "type": "object", + "required": ["target_arn", "replicated_count", "replicated_size_bytes", "failed_count", "failed_size_bytes", "latency", "bandwidth", "extensions"], + "properties": { + "target_arn": { "type": "string" }, + "replicated_count": { "type": "integer", "minimum": 0 }, + "replicated_size_bytes": { "type": "integer", "minimum": 0 }, + "failed_count": { "type": "integer", "minimum": 0 }, + "failed_size_bytes": { "type": "integer", "minimum": 0 }, + "latency": { + "type": "object", + "required": ["average_ms", "current_ms", "maximum_ms", "scope"], + "properties": { + "average_ms": { "type": "number", "minimum": 0 }, + "current_ms": { "type": "number", "minimum": 0 }, + "maximum_ms": { "type": "number", "minimum": 0 }, + "scope": { "type": "string", "minLength": 1 } + } + }, + "bandwidth": { + "type": "object", + "required": ["limit_bytes_per_sec", "current_bytes_per_sec", "scope"], + "properties": { + "limit_bytes_per_sec": { "type": "integer", "minimum": 0 }, + "current_bytes_per_sec": { "type": "number", "minimum": 0 }, + "scope": { "type": "string", "minLength": 1 } + } + }, + "extensions": { "type": "object" } + } + } + }, + "extensions": { "type": "object" } + } + }, + "replicationMrfData": { + "type": "object", + "required": ["operation", "bucket", "runtime_stats_available", "cluster", "failed", "queue", "durable_backlog", "per_object_entries_available", "per_target_durable_entries_available", "targets", "extensions"], + "properties": { + "operation": { "const": "mrf" }, + "bucket": { "type": "string", "minLength": 1 }, + "runtime_stats_available": { "type": "boolean" }, + "cluster": { "$ref": "#/definitions/replicationCluster" }, + "failed": { "$ref": "#/definitions/replicationBacklog" }, + "queue": { "$ref": "#/definitions/replicationBacklog" }, + "durable_backlog": { + "type": "object", + "required": ["available", "count", "size_bytes"], + "properties": { + "available": { "type": "boolean" }, + "count": { "type": "integer", "minimum": 0 }, + "size_bytes": { "type": "integer", "minimum": 0 } + } + }, + "per_object_entries_available": { "type": "boolean" }, + "per_target_durable_entries_available": { "type": "boolean" }, + "targets": { + "type": "array", + "items": { + "type": "object", + "required": ["target_arn", "failed_count", "failed_size_bytes", "observation_scope", "extensions"], + "properties": { + "target_arn": { "type": "string" }, + "failed_count": { "type": "integer", "minimum": 0 }, + "failed_size_bytes": { "type": "integer", "minimum": 0 }, + "observation_scope": { "type": "string", "minLength": 1 }, + "extensions": { "type": "object" } + } + } }, - "scan": { "$ref": "#/definitions/replicationDiffScan" }, "extensions": { "type": "object" } } },