From 938f7296f9ad14f320420f419809014f54140df3 Mon Sep 17 00:00:00 2001 From: cxymds Date: Fri, 24 Jul 2026 23:33:45 +0800 Subject: [PATCH] fix(admin): report truthful diagnostic capabilities (#5205) --- rustfs/src/admin/handlers/diagnostics.rs | 170 +++++++++++++- rustfs/src/admin/handlers/site_replication.rs | 83 ++++--- rustfs/src/admin/handlers/system.rs | 132 +++++++++++ .../tests/admin_diagnostic_capability_e2e.rs | 218 ++++++++++++++++++ 4 files changed, 560 insertions(+), 43 deletions(-) create mode 100644 rustfs/tests/admin_diagnostic_capability_e2e.rs diff --git a/rustfs/src/admin/handlers/diagnostics.rs b/rustfs/src/admin/handlers/diagnostics.rs index 0564d21b0..523df19db 100644 --- a/rustfs/src/admin/handlers/diagnostics.rs +++ b/rustfs/src/admin/handlers/diagnostics.rs @@ -30,7 +30,7 @@ use crate::auth::{check_key_valid, get_session_token}; use crate::server::{ADMIN_PREFIX, RemoteAddr}; use bytes::Bytes; use futures::{Stream, StreamExt}; -use http::{HeaderMap, HeaderValue, Uri}; +use http::{HeaderMap, HeaderValue, Uri, header::CONTENT_LENGTH}; use hyper::{Method, StatusCode}; use matchit::Params; use rustfs_lock::{LockMode, ObjectKey, get_global_lock_manager}; @@ -42,12 +42,16 @@ use serde::{Deserialize, Serialize}; use std::pin::Pin; use std::task::{Context, Poll}; use std::time::{Duration, SystemTime}; -use tokio::sync::mpsc; +use tokio::sync::{Semaphore, SemaphorePermit, mpsc}; use tokio_stream::wrappers::ReceiverStream; use tracing::warn; const CONTENT_TYPE_JSON: &str = "application/json"; const CONTENT_TYPE_NDJSON: &str = "application/x-ndjson"; +pub(crate) const CLIENT_DEVNULL_MAX_BYTES: u64 = 1024 * 1024 * 1024; +pub(crate) const CLIENT_DEVNULL_MAX_DURATION: Duration = Duration::from_secs(30); +pub(crate) const CLIENT_DEVNULL_MAX_CONCURRENCY: usize = 4; +static CLIENT_DEVNULL_ADMISSION: Semaphore = Semaphore::const_new(CLIENT_DEVNULL_MAX_CONCURRENCY); /// Cap on how many locks a single `top/locks` response enumerates, matching the /// MinIO default page size and bounding response size on busy clusters. @@ -811,18 +815,75 @@ impl Operation for SpeedtestHandler { /// number (mirrors MinIO's `ClientDevNull`). pub struct SpeedtestClientDevnullHandler {} +fn validate_client_devnull_content_length(headers: &HeaderMap) -> S3Result<()> { + let Some(content_length) = headers.get(CONTENT_LENGTH) else { + return Ok(()); + }; + let content_length = content_length + .to_str() + .ok() + .and_then(|value| value.parse::().ok()) + .ok_or_else(|| s3_error!(InvalidRequest, "invalid Content-Length for client devnull stream"))?; + if content_length > CLIENT_DEVNULL_MAX_BYTES { + return Err(s3_error!( + EntityTooLarge, + "client devnull stream exceeds the {}-byte limit", + CLIENT_DEVNULL_MAX_BYTES + )); + } + Ok(()) +} + +fn acquire_client_devnull_permit() -> S3Result> { + CLIENT_DEVNULL_ADMISSION.try_acquire().map_err(|_| { + s3_error!( + SlowDown, + "client devnull concurrency limit of {} is exhausted", + CLIENT_DEVNULL_MAX_CONCURRENCY + ) + }) +} + +async fn drain_client_devnull(input: Body, max_bytes: u64, max_duration: Duration) -> S3Result { + tokio::time::timeout(max_duration, async move { + let mut input = input; + let mut total = 0u64; + while let Some(chunk) = input.next().await { + let chunk = chunk.map_err(|e| s3_error!(InvalidRequest, "failed to read devnull stream: {}", e))?; + let chunk_len = u64::try_from(chunk.len()) + .map_err(|_| s3_error!(InternalError, "devnull stream chunk length exceeds supported range"))?; + total = total + .checked_add(chunk_len) + .ok_or_else(|| s3_error!(EntityTooLarge, "client devnull stream exceeds the configured byte limit"))?; + if total > max_bytes { + return Err(s3_error!(EntityTooLarge, "client devnull stream exceeds the {}-byte limit", max_bytes)); + } + } + Ok(total) + }) + .await + .map_err(|_| { + s3_error!( + RequestTimeout, + "client devnull stream exceeded the {}-second limit", + max_duration.as_secs() + ) + })? +} + +async fn drain_client_devnull_with_production_limits(input: Body) -> S3Result { + drain_client_devnull(input, CLIENT_DEVNULL_MAX_BYTES, CLIENT_DEVNULL_MAX_DURATION).await +} + #[async_trait::async_trait] impl Operation for SpeedtestClientDevnullHandler { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { authorize(&req, AdminAction::HealthInfoAdminAction).await?; + validate_client_devnull_content_length(&req.headers)?; + let _permit = acquire_client_devnull_permit()?; let started = std::time::Instant::now(); - let mut input = req.input; - let mut total: u64 = 0; - while let Some(chunk) = input.next().await { - let chunk = chunk.map_err(|e| s3_error!(InvalidRequest, "failed to read devnull stream: {}", e))?; - total += chunk.len() as u64; - } + let total = drain_client_devnull_with_production_limits(req.input).await?; let elapsed = started.elapsed(); let response = SpeedtestResponse { @@ -918,6 +979,99 @@ mod tests { assert_eq!(err.code(), &S3ErrorCode::AccessDenied); } + #[tokio::test] + async fn client_devnull_enforces_byte_limit_while_streaming() { + let err = drain_client_devnull(Body::from(vec![0u8; 5]), 4, Duration::from_secs(1)) + .await + .expect_err("stream over the byte limit must fail"); + + assert_eq!(err.code(), &S3ErrorCode::EntityTooLarge); + } + + #[tokio::test] + async fn client_devnull_enforces_duration_limit() { + let (tx, rx) = mpsc::channel::>(1); + let stream: DynByteStream = Box::pin(ByteChannelStream { + inner: ReceiverStream::new(rx), + }); + let body = Body::from(stream); + + let err = drain_client_devnull(body, 4, Duration::ZERO) + .await + .expect_err("stalled stream must time out"); + assert_eq!(err.code(), &S3ErrorCode::RequestTimeout); + assert!( + tx.send(Ok(Bytes::from_static(b"late"))).await.is_err(), + "timeout must drop the request body and cancel its upstream receiver" + ); + } + + #[tokio::test(start_paused = true)] + async fn client_devnull_production_limits_enforce_advertised_timeout() { + let (tx, rx) = mpsc::channel::>(1); + let stream: DynByteStream = Box::pin(ByteChannelStream { + inner: ReceiverStream::new(rx), + }); + let task = tokio::spawn(drain_client_devnull_with_production_limits(Body::from(stream))); + + tokio::task::yield_now().await; + tokio::time::advance(CLIENT_DEVNULL_MAX_DURATION + Duration::from_millis(1)).await; + + let err = task + .await + .expect("production drain task must not panic") + .expect_err("stalled production stream must time out at the advertised limit"); + assert_eq!(err.code(), &S3ErrorCode::RequestTimeout); + assert!( + tx.send(Ok(Bytes::from_static(b"late"))).await.is_err(), + "production timeout must drop the request body and cancel its upstream receiver" + ); + } + + #[tokio::test] + async fn client_devnull_reports_exact_bounded_byte_count() { + let total = drain_client_devnull(Body::from(vec![0u8; 4]), 4, Duration::from_secs(1)) + .await + .expect("stream at the byte limit should succeed"); + + assert_eq!(total, 4); + } + + #[test] + fn client_devnull_rejects_invalid_content_length() { + let mut headers = HeaderMap::new(); + headers.insert(CONTENT_LENGTH, HeaderValue::from_static("not-a-number")); + + let err = validate_client_devnull_content_length(&headers) + .expect_err("malformed content length must fail before reading the stream"); + assert_eq!(err.code(), &S3ErrorCode::InvalidRequest); + } + + #[test] + fn client_devnull_rejects_oversized_content_length() { + let mut headers = HeaderMap::new(); + headers.insert( + CONTENT_LENGTH, + HeaderValue::from_str(&(CLIENT_DEVNULL_MAX_BYTES + 1).to_string()).expect("valid header"), + ); + + let err = validate_client_devnull_content_length(&headers) + .expect_err("oversized content length must fail before reading the stream"); + assert_eq!(err.code(), &S3ErrorCode::EntityTooLarge); + } + + #[test] + fn client_devnull_admission_fails_fast_and_recovers() { + let permits = (0..CLIENT_DEVNULL_MAX_CONCURRENCY) + .map(|_| acquire_client_devnull_permit().expect("admission slot")) + .collect::>(); + let err = acquire_client_devnull_permit().expect_err("exhausted admission must fail"); + assert_eq!(err.code(), &S3ErrorCode::SlowDown); + + drop(permits); + let _permit = acquire_client_devnull_permit().expect("released admission slots must be reusable"); + } + #[test] fn speedtest_kind_routing() { assert_eq!(speedtest_kind_from_path("/rustfs/admin/v3/speedtest"), SpeedtestKind::Object); diff --git a/rustfs/src/admin/handlers/site_replication.rs b/rustfs/src/admin/handlers/site_replication.rs index f5a00708a..ab251be94 100644 --- a/rustfs/src/admin/handlers/site_replication.rs +++ b/rustfs/src/admin/handlers/site_replication.rs @@ -97,7 +97,7 @@ use std::net::{IpAddr, SocketAddr}; #[cfg(test)] use std::sync::Arc; use std::sync::{LazyLock, Mutex as StdMutex}; -use std::time::{Duration, Instant}; +use std::time::Duration; use time::OffsetDateTime; use tokio::sync::{Mutex, RwLock}; use tracing::{info, warn}; @@ -117,8 +117,6 @@ const SITE_REPL_RESYNC_CANCEL: &str = "cancel"; const SITE_REPL_RESYNC_STATUS: &str = "status"; const SITE_REPL_RESYNC_DEFAULT_PAGE_SIZE: usize = 100; const SITE_REPL_RESYNC_MAX_PAGE_SIZE: usize = 1000; -const SITE_REPL_MIN_NETPERF_DURATION: Duration = Duration::from_secs(1); -const SITE_REPL_MAX_NETPERF_DURATION: Duration = Duration::from_secs(30); const SITE_REPLICATION_PEER_REQUEST_TIMEOUT: Duration = Duration::from_secs(10); const SITE_REPLICATION_PEER_CONNECT_TIMEOUT: Duration = Duration::from_secs(3); const SITE_REPLICATION_PEER_ERROR_DETAIL_LIMIT: usize = 256; @@ -767,14 +765,6 @@ fn go_gob_site_netperf_response(value: &SiteNetPerfNodeResult) -> S3Response<(St S3Response::new((StatusCode::OK, Body::from(data))) } -fn site_repl_netperf_duration(uri: &Uri) -> Duration { - query_pairs(uri) - .get("duration") - .and_then(|value| rustfs_madmin::utils::parse_duration(value).ok()) - .unwrap_or(SITE_REPL_MIN_NETPERF_DURATION) - .clamp(SITE_REPL_MIN_NETPERF_DURATION, SITE_REPL_MAX_NETPERF_DURATION) -} - fn encode_go_gob_site_netperf_node_result(value: &SiteNetPerfNodeResult) -> Vec { let mut data = GO_GOB_SITE_NETPERF_SCHEMA.to_vec(); let mut payload = Vec::new(); @@ -6985,26 +6975,24 @@ impl Operation for SiteReplicationDevNullHandler { pub struct SiteReplicationNetPerfHandler {} +fn unsupported_site_netperf_result(endpoint: String) -> SiteNetPerfNodeResult { + SiteNetPerfNodeResult { + endpoint, + tx: 0, + tx_total_duration_ns: 0, + rx: 0, + rx_total_duration_ns: 0, + total_conn: 0, + error: "site-replication netperf is unsupported because RustFS does not perform peer traffic".to_string(), + } +} + #[async_trait::async_trait] impl Operation for SiteReplicationNetPerfHandler { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { validate_site_replication_admin_request(&req, AdminAction::SiteReplicationOperationAction).await?; - let duration = site_repl_netperf_duration(&req.uri); - let endpoint = request_endpoint(&req.uri, &req.headers); - let started_at = Instant::now(); - let body = read_plain_admin_body(req.input).await?; - let elapsed = started_at.elapsed().max(duration); - - Ok(go_gob_site_netperf_response(&SiteNetPerfNodeResult { - endpoint, - tx: body.len() as u64, - tx_total_duration_ns: elapsed.as_nanos() as i64, - rx: body.len() as u64, - rx_total_duration_ns: elapsed.as_nanos() as i64, - total_conn: 1, - error: String::new(), - })) + Ok(go_gob_site_netperf_response(&unsupported_site_netperf_result(endpoint))) } } @@ -11823,16 +11811,14 @@ mod tests { } #[test] - fn test_site_repl_netperf_duration_is_bounded() { - let default_uri = Uri::from_static("/rustfs/admin/v3/site-replication/netperf"); - let too_short_uri = Uri::from_static("/rustfs/admin/v3/site-replication/netperf?duration=0s"); - let too_long_uri = Uri::from_static("/rustfs/admin/v3/site-replication/netperf?duration=60s"); - let valid_uri = Uri::from_static("/rustfs/admin/v3/site-replication/netperf?duration=5s"); + fn test_site_repl_netperf_reports_unsupported_without_measurements() { + let result = unsupported_site_netperf_result("https://peer.example.com".to_string()); - assert_eq!(site_repl_netperf_duration(&default_uri), SITE_REPL_MIN_NETPERF_DURATION); - assert_eq!(site_repl_netperf_duration(&too_short_uri), SITE_REPL_MIN_NETPERF_DURATION); - assert_eq!(site_repl_netperf_duration(&too_long_uri), SITE_REPL_MAX_NETPERF_DURATION); - assert_eq!(site_repl_netperf_duration(&valid_uri), Duration::from_secs(5)); + assert_eq!(result.endpoint, "https://peer.example.com"); + assert_eq!(result.tx, 0); + assert_eq!(result.rx, 0); + assert_eq!(result.total_conn, 0); + assert!(result.error.contains("unsupported")); } #[test] @@ -11863,6 +11849,33 @@ mod tests { assert_eq!(data, expected); } + #[test] + fn test_gob_site_netperf_unsupported_error_matches_go_encoding() { + let data = + encode_go_gob_site_netperf_node_result(&unsupported_site_netperf_result("https://peer.example.com".to_string())); + + // Generated independently with Go's encoding/gob Encoder from the + // MinIO-compatible SiteNetPerfNodeResult shape. This specifically + // covers the field delta from Endpoint to Error when all counters are zero. + let expected: &[u8] = &[ + 0x7d, 0x7f, 0x03, 0x01, 0x01, 0x15, 0x53, 0x69, 0x74, 0x65, 0x4e, 0x65, 0x74, 0x50, 0x65, 0x72, 0x66, 0x4e, 0x6f, + 0x64, 0x65, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x01, 0xff, 0x80, 0x00, 0x01, 0x07, 0x01, 0x08, 0x45, 0x6e, 0x64, + 0x70, 0x6f, 0x69, 0x6e, 0x74, 0x01, 0x0c, 0x00, 0x01, 0x02, 0x54, 0x58, 0x01, 0x06, 0x00, 0x01, 0x0f, 0x54, 0x58, + 0x54, 0x6f, 0x74, 0x61, 0x6c, 0x44, 0x75, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x01, 0x04, 0x00, 0x01, 0x02, 0x52, + 0x58, 0x01, 0x06, 0x00, 0x01, 0x0f, 0x52, 0x58, 0x54, 0x6f, 0x74, 0x61, 0x6c, 0x44, 0x75, 0x72, 0x61, 0x74, 0x69, + 0x6f, 0x6e, 0x01, 0x04, 0x00, 0x01, 0x09, 0x54, 0x6f, 0x74, 0x61, 0x6c, 0x43, 0x6f, 0x6e, 0x6e, 0x01, 0x06, 0x00, + 0x01, 0x05, 0x45, 0x72, 0x72, 0x6f, 0x72, 0x01, 0x0c, 0x00, 0x00, 0x00, 0x73, 0xff, 0x80, 0x01, 0x18, 0x68, 0x74, + 0x74, 0x70, 0x73, 0x3a, 0x2f, 0x2f, 0x70, 0x65, 0x65, 0x72, 0x2e, 0x65, 0x78, 0x61, 0x6d, 0x70, 0x6c, 0x65, 0x2e, + 0x63, 0x6f, 0x6d, 0x06, 0x54, 0x73, 0x69, 0x74, 0x65, 0x2d, 0x72, 0x65, 0x70, 0x6c, 0x69, 0x63, 0x61, 0x74, 0x69, + 0x6f, 0x6e, 0x20, 0x6e, 0x65, 0x74, 0x70, 0x65, 0x72, 0x66, 0x20, 0x69, 0x73, 0x20, 0x75, 0x6e, 0x73, 0x75, 0x70, + 0x70, 0x6f, 0x72, 0x74, 0x65, 0x64, 0x20, 0x62, 0x65, 0x63, 0x61, 0x75, 0x73, 0x65, 0x20, 0x52, 0x75, 0x73, 0x74, + 0x46, 0x53, 0x20, 0x64, 0x6f, 0x65, 0x73, 0x20, 0x6e, 0x6f, 0x74, 0x20, 0x70, 0x65, 0x72, 0x66, 0x6f, 0x72, 0x6d, + 0x20, 0x70, 0x65, 0x65, 0x72, 0x20, 0x74, 0x72, 0x61, 0x66, 0x66, 0x69, 0x63, 0x00, + ]; + + assert_eq!(data, expected); + } + #[test] fn test_group_info_with_empty_members_still_requires_group_upsert() { let update = rustfs_madmin::GroupAddRemove { diff --git a/rustfs/src/admin/handlers/system.rs b/rustfs/src/admin/handlers/system.rs index bf03f61fd..2eb2014f9 100644 --- a/rustfs/src/admin/handlers/system.rs +++ b/rustfs/src/admin/handlers/system.rs @@ -636,6 +636,7 @@ pub struct RuntimeCapabilitiesSummary { #[derive(Debug, Clone, PartialEq, Eq, Serialize)] pub struct RuntimeCapabilitiesResponse { pub summary: RuntimeCapabilitiesSummary, + pub diagnostic_probes: DiagnosticProbeCapabilities, pub storage_classes: StorageClassCapabilities, pub cluster_snapshot_path: String, pub cluster_snapshot_summary: Option, @@ -645,6 +646,99 @@ pub struct RuntimeCapabilitiesResponse { pub topology_status: CapabilityStatus, } +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub struct DiagnosticProbeCapabilities { + pub contract_version: u32, + pub health_info: DiagnosticProbeCapability, + pub inspect_archive: DiagnosticProbeCapability, + pub observed_drive_metrics: DiagnosticProbeCapability, + pub client_devnull: DiagnosticProbeCapability, + pub object_speedtest: DiagnosticProbeCapability, + pub inter_node_netperf: DiagnosticProbeCapability, + pub site_speedtest: DiagnosticProbeCapability, + pub site_replication_netperf: DiagnosticProbeCapability, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub struct DiagnosticProbeCapability { + pub status: CapabilityStatus, + pub mode: &'static str, + #[serde(skip_serializing_if = "Option::is_none")] + pub route: Option<&'static str>, + #[serde(skip_serializing_if = "Option::is_none")] + pub max_bytes: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub max_duration_secs: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub max_concurrency: Option, +} + +impl DiagnosticProbeCapabilities { + fn current() -> Self { + let unsupported = |reason, route| DiagnosticProbeCapability { + status: CapabilityStatus::unsupported().with_reason(reason), + mode: "unsupported", + route, + max_bytes: None, + max_duration_secs: None, + max_concurrency: None, + }; + + Self { + contract_version: 1, + health_info: DiagnosticProbeCapability { + status: CapabilityStatus::supported().with_reason("reports local host telemetry and observed storage state"), + mode: "system_observation", + route: Some("/rustfs/admin/v3/healthinfo"), + max_bytes: None, + max_duration_secs: None, + max_concurrency: None, + }, + inspect_archive: unsupported( + "a bounded encrypted diagnostic archive is not implemented; v3 inspect-data returns object bytes", + None, + ), + observed_drive_metrics: DiagnosticProbeCapability { + status: CapabilityStatus::supported() + .with_reason("reports observed StorageInfo throughput and latency, not an active benchmark"), + mode: "observed_storage_metrics", + route: Some("/rustfs/admin/v3/speedtest/drive"), + max_bytes: None, + max_duration_secs: None, + max_concurrency: None, + }, + client_devnull: DiagnosticProbeCapability { + status: CapabilityStatus::supported().with_reason("actively drains a bounded client upload"), + mode: "active_upload_drain", + route: Some("/rustfs/admin/v3/speedtest/client/devnull"), + max_bytes: Some(super::diagnostics::CLIENT_DEVNULL_MAX_BYTES), + max_duration_secs: Some(super::diagnostics::CLIENT_DEVNULL_MAX_DURATION.as_secs()), + max_concurrency: Some(super::diagnostics::CLIENT_DEVNULL_MAX_CONCURRENCY), + }, + object_speedtest: unsupported( + "object PUT/GET benchmark harness is not implemented", + Some("/rustfs/admin/v3/speedtest/object"), + ), + inter_node_netperf: unsupported( + "inter-node traffic benchmark harness is not implemented", + Some("/rustfs/admin/v3/speedtest/net"), + ), + site_speedtest: unsupported( + "site traffic benchmark harness is not implemented", + Some("/rustfs/admin/v3/speedtest/site"), + ), + site_replication_netperf: DiagnosticProbeCapability { + status: CapabilityStatus::unsupported().with_reason("site-replication netperf does not perform peer traffic"), + mode: "unsupported", + route: Some("/rustfs/admin/v3/site-replication/netperf"), + max_bytes: None, + max_duration_secs: None, + max_concurrency: None, + }, + } + } +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize)] pub struct StorageClassCapabilities { pub contract_version: u32, @@ -690,6 +784,7 @@ pub(crate) async fn build_runtime_capabilities_response() Ok(RuntimeCapabilitiesResponse { summary, + diagnostic_probes: DiagnosticProbeCapabilities::current(), storage_classes: StorageClassCapabilities::current(), cluster_snapshot_path: usecase.cluster_snapshot_route().to_string(), cluster_snapshot_summary: cluster_snapshot_discovery.summary, @@ -953,6 +1048,33 @@ mod tests { assert_eq!(response.summary.site_replication_info.state, CapabilityState::Supported); assert_eq!(response.summary.site_replication_edit.state, CapabilityState::Supported); assert_eq!(response.summary.site_replication_resync.state, CapabilityState::Supported); + assert_eq!(response.diagnostic_probes.contract_version, 1); + assert_eq!(response.diagnostic_probes.health_info.status.state, CapabilityState::Supported); + assert_eq!(response.diagnostic_probes.observed_drive_metrics.mode, "observed_storage_metrics"); + assert_eq!(response.diagnostic_probes.client_devnull.status.state, CapabilityState::Supported); + assert_eq!( + response.diagnostic_probes.client_devnull.max_bytes, + Some(super::super::diagnostics::CLIENT_DEVNULL_MAX_BYTES) + ); + assert_eq!( + response.diagnostic_probes.client_devnull.max_duration_secs, + Some(super::super::diagnostics::CLIENT_DEVNULL_MAX_DURATION.as_secs()) + ); + assert_eq!( + response.diagnostic_probes.client_devnull.max_concurrency, + Some(super::super::diagnostics::CLIENT_DEVNULL_MAX_CONCURRENCY) + ); + for probe in [ + &response.diagnostic_probes.inspect_archive, + &response.diagnostic_probes.object_speedtest, + &response.diagnostic_probes.inter_node_netperf, + &response.diagnostic_probes.site_speedtest, + &response.diagnostic_probes.site_replication_netperf, + ] { + assert_eq!(probe.status.state, CapabilityState::Unsupported); + assert_eq!(probe.mode, "unsupported"); + assert!(probe.status.reason.is_some()); + } assert_eq!(response.storage_classes.contract_version, 1); assert_eq!(response.storage_classes.supported_write_classes, ["STANDARD", "REDUCED_REDUNDANCY"]); assert_eq!(response.storage_classes.unsupported_write_error, "InvalidStorageClass"); @@ -962,6 +1084,16 @@ mod tests { assert_eq!(value["summary"]["site_replication_info"]["state"], "supported"); assert_eq!(value["summary"]["site_replication_edit"]["state"], "supported"); assert_eq!(value["summary"]["site_replication_resync"]["state"], "supported"); + assert_eq!(value["diagnostic_probes"]["client_devnull"]["status"]["state"], "supported"); + assert_eq!( + value["diagnostic_probes"]["client_devnull"]["max_bytes"], + super::super::diagnostics::CLIENT_DEVNULL_MAX_BYTES + ); + assert_eq!( + value["diagnostic_probes"]["client_devnull"]["max_concurrency"], + super::super::diagnostics::CLIENT_DEVNULL_MAX_CONCURRENCY + ); + assert_eq!(value["diagnostic_probes"]["site_replication_netperf"]["status"]["state"], "unsupported"); assert_eq!(value["storage_classes"]["contract_version"], 1); assert_eq!( value["storage_classes"]["supported_write_classes"], diff --git a/rustfs/tests/admin_diagnostic_capability_e2e.rs b/rustfs/tests/admin_diagnostic_capability_e2e.rs new file mode 100644 index 000000000..f639802b1 --- /dev/null +++ b/rustfs/tests/admin_diagnostic_capability_e2e.rs @@ -0,0 +1,218 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 + +use bytes::Bytes; +use chrono::Utc; +use futures::stream; +use hmac::{Hmac, KeyInit, Mac}; +use reqwest::{Client, Method, Response}; +use rustfs::embedded::{RustFSServerBuilder, find_available_port}; +use sha2::{Digest, Sha256}; +use std::time::Duration; + +type HmacSha256 = Hmac; + +fn hex(bytes: impl AsRef<[u8]>) -> String { + bytes.as_ref().iter().map(|byte| format!("{byte:02x}")).collect() +} + +fn sha256_hex(bytes: &[u8]) -> String { + hex(Sha256::digest(bytes)) +} + +fn hmac(key: &[u8], value: &str) -> Vec { + let mut mac = HmacSha256::new_from_slice(key).expect("HMAC accepts arbitrary key lengths"); + mac.update(value.as_bytes()); + mac.finalize().into_bytes().to_vec() +} + +fn signed_admin_request( + client: &Client, + endpoint: &str, + method: Method, + path: &str, + access_key: &str, + secret_key: &str, + payload_hash: &str, +) -> reqwest::RequestBuilder { + let host = endpoint + .strip_prefix("http://") + .or_else(|| endpoint.strip_prefix("https://")) + .expect("embedded endpoint scheme"); + let now = Utc::now(); + let amz_date = now.format("%Y%m%dT%H%M%SZ").to_string(); + let date = now.format("%Y%m%d").to_string(); + let canonical_headers = format!("host:{host}\nx-amz-content-sha256:{payload_hash}\nx-amz-date:{amz_date}\n"); + let signed_headers = "host;x-amz-content-sha256;x-amz-date"; + let canonical_request = format!("{}\n{path}\n\n{canonical_headers}\n{signed_headers}\n{payload_hash}", method.as_str()); + let scope = format!("{date}/us-east-1/s3/aws4_request"); + let string_to_sign = format!("AWS4-HMAC-SHA256\n{amz_date}\n{scope}\n{}", sha256_hex(canonical_request.as_bytes())); + let date_key = hmac(format!("AWS4{secret_key}").as_bytes(), &date); + let region_key = hmac(&date_key, "us-east-1"); + let service_key = hmac(®ion_key, "s3"); + let signing_key = hmac(&service_key, "aws4_request"); + let signature = hex(hmac(&signing_key, &string_to_sign)); + let authorization = + format!("AWS4-HMAC-SHA256 Credential={access_key}/{scope}, SignedHeaders={signed_headers}, Signature={signature}"); + + client + .request(method, format!("{endpoint}{path}")) + .header("host", host) + .header("x-amz-content-sha256", payload_hash) + .header("x-amz-date", amz_date) + .header("authorization", authorization) +} + +async fn signed_bytes_request( + client: &Client, + endpoint: &str, + method: Method, + path: &str, + access_key: &str, + secret_key: &str, + body: &'static [u8], +) -> Response { + signed_admin_request(client, endpoint, method, path, access_key, secret_key, &sha256_hex(body)) + .body(body) + .send() + .await + .expect("signed admin request") +} + +#[tokio::test] +async fn diagnostic_handlers_enforce_advertised_runtime_contract() { + let port = match find_available_port() { + Ok(port) => port, + Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return, + Err(err) => panic!("find free port: {err}"), + }; + let server = RustFSServerBuilder::new() + .address(format!("127.0.0.1:{port}")) + .access_key("diagnostic-e2e-access") + .secret_key("diagnostic-e2e-secret") + .build() + .await + .expect("start embedded server"); + let endpoint = server.endpoint(); + let client = Client::builder() + .timeout(Duration::from_secs(5)) + .build() + .expect("HTTP client"); + + let response = signed_bytes_request( + &client, + &endpoint, + Method::POST, + "/rustfs/admin/v3/speedtest/client/devnull", + server.access_key(), + server.secret_key(), + b"bounded", + ) + .await; + assert!(response.status().is_success()); + let value: serde_json::Value = response.json().await.expect("client devnull JSON"); + assert_eq!(value["kind"], "client-devnull"); + assert_eq!(value["measured"], true); + assert_eq!(value["rx_bytes"], 7); + + let oversized = signed_admin_request( + &client, + &endpoint, + Method::POST, + "/rustfs/admin/v3/speedtest/client/devnull", + server.access_key(), + server.secret_key(), + &sha256_hex(b""), + ) + .header("content-length", 1024_u64 * 1024 * 1024 + 1) + .body(reqwest::Body::wrap_stream(stream::pending::>())) + .send() + .await + .expect("oversized request must receive an early response"); + assert_eq!(oversized.status(), reqwest::StatusCode::BAD_REQUEST); + assert!( + oversized + .text() + .await + .expect("oversized error body") + .contains("EntityTooLarge") + ); + + let stalled_client = Client::builder() + .timeout(Duration::from_secs(60)) + .build() + .expect("stalled HTTP client"); + let mut stalled = Vec::new(); + for _ in 0..4 { + let client = stalled_client.clone(); + let endpoint = endpoint.clone(); + let access_key = server.access_key().to_string(); + let secret_key = server.secret_key().to_string(); + stalled.push(tokio::spawn(async move { + signed_admin_request( + &client, + &endpoint, + Method::POST, + "/rustfs/admin/v3/speedtest/client/devnull", + &access_key, + &secret_key, + &sha256_hex(b"x"), + ) + .header("content-length", 1) + .body(reqwest::Body::wrap_stream(stream::pending::>())) + .send() + .await + })); + } + tokio::time::timeout(Duration::from_secs(10), async { + loop { + let probe = signed_bytes_request( + &client, + &endpoint, + Method::POST, + "/rustfs/admin/v3/speedtest/client/devnull", + server.access_key(), + server.secret_key(), + b"x", + ) + .await; + let status = probe.status(); + let body = probe.text().await.expect("admission probe body"); + if status == reqwest::StatusCode::SERVICE_UNAVAILABLE { + assert!(body.contains("SlowDown")); + break; + } + assert!(status.is_success(), "unexpected admission probe status {status}: {body}"); + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("stalled requests must occupy every admission slot"); + for request in stalled { + request.abort(); + } + + let netperf = signed_bytes_request( + &client, + &endpoint, + Method::POST, + "/rustfs/admin/v3/site-replication/netperf", + server.access_key(), + server.secret_key(), + b"", + ) + .await; + assert!(netperf.status().is_success()); + let gob = netperf.bytes().await.expect("site netperf gob"); + assert!( + gob.windows(b"site-replication netperf is unsupported because RustFS does not perform peer traffic".len()) + .any(|window| window == b"site-replication netperf is unsupported because RustFS does not perform peer traffic") + ); + + server.shutdown().await; +}