From 46775ae019a2069e5607ae50cba488cb3125a939 Mon Sep 17 00:00:00 2001 From: houseme Date: Tue, 23 Jun 2026 18:10:49 +0800 Subject: [PATCH] feat: harden runtime capability snapshots (#3784) * feat(admin): expose runtime capability snapshots * feat(runtime): refine workload admission snapshots * test(ci): align architecture migration checks * build(deps): bump memmap2 for RUSTSEC-2026-0186 * build(deps): refresh cargo deny lockfile --- Cargo.lock | 30 +- Cargo.toml | 2 +- crates/concurrency/Cargo.toml | 2 + crates/concurrency/src/workload.rs | 50 ++- crates/ecstore/src/cluster/mod.rs | 3 + deny.toml | 8 - rustfs/src/admin/handlers/extensions.rs | 75 +++- rustfs/src/admin/handlers/system.rs | 334 ++++++++++++++++++ rustfs/src/admin/route_policy.rs | 6 + rustfs/src/admin/route_registration_test.rs | 2 + rustfs/src/runtime_capabilities.rs | 68 +++- rustfs/src/workload_admission.rs | 137 ++++--- scripts/check_architecture_migration_rules.sh | 20 +- 13 files changed, 646 insertions(+), 91 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 7be1a185b..5c8c22440 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2340,9 +2340,9 @@ dependencies = [ [[package]] name = "crypto-bigint" -version = "0.7.4" +version = "0.7.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "97bb4a855e3b10f84c4e7e895a7de01db7f9a7b7eb7f73ed9773fd52ac686451" +checksum = "1a52aa3fcda4e6302a9f48734f234d35d4721b96f8fe07d073f07ce9df4f0271" dependencies = [ "cpubits", "ctutils", @@ -2382,7 +2382,7 @@ version = "0.7.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3633a51a39c69ebbaa4feaa694bd83d241e4093901c84a0963b19d9bb3f0cf8f" dependencies = [ - "crypto-bigint 0.7.4", + "crypto-bigint 0.7.5", "rand_core 0.10.1", ] @@ -3847,7 +3847,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "102d3643d30dd8b559613c5cced68317199597fffb278cdc88daa2ef7fafc935" dependencies = [ "base16ct 1.0.0", - "crypto-bigint 0.7.4", + "crypto-bigint 0.7.5", "crypto-common 0.2.2", "digest 0.11.3", "ff 0.14.0", @@ -6111,9 +6111,9 @@ dependencies = [ [[package]] name = "memmap2" -version = "0.9.10" +version = "0.9.11" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "714098028fe011992e1c3962653c96b2d578c4b4bce9036e15ff220319b1e0e3" +checksum = "d1219ed1b7f229ee7104d281dd01d6802fe28bb6e95d292942c4daacdeb798c0" dependencies = [ "libc", ] @@ -7834,7 +7834,7 @@ version = "0.14.0-rc.12" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8675564771a62f69a0af716b03e89b917b963c7b173b5855575e84fd4f605ca0" dependencies = [ - "crypto-bigint 0.7.4", + "crypto-bigint 0.7.5", "crypto-common 0.2.2", "ff 0.14.0", "rand_core 0.10.1", @@ -8805,7 +8805,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "30b2aa4ba0d89f73d1e332df05be0eeab8840351c36ca5654341dfdb57bb3caf" dependencies = [ "const-oid 0.10.2", - "crypto-bigint 0.7.4", + "crypto-bigint 0.7.5", "crypto-primes", "digest 0.11.3", "pkcs1 0.8.0-rc.4", @@ -8883,7 +8883,7 @@ dependencies = [ "bytes", "cbc 0.2.1", "cipher 0.5.2", - "crypto-bigint 0.7.4", + "crypto-bigint 0.7.5", "ctr", "curve25519-dalek 5.0.0-rc.0", "data-encoding", @@ -9227,6 +9227,8 @@ version = "1.0.0-beta.8" dependencies = [ "rustfs-io-core", "rustfs-io-metrics", + "serde", + "serde_json", "thiserror 2.0.18", "tokio", "tokio-util", @@ -9326,7 +9328,7 @@ dependencies = [ "lazy_static", "libc", "md-5 0.11.0", - "memmap2 0.9.10", + "memmap2 0.9.11", "metrics", "num_cpus", "opentelemetry", @@ -9513,7 +9515,7 @@ name = "rustfs-io-core" version = "1.0.0-beta.8" dependencies = [ "bytes", - "memmap2 0.9.10", + "memmap2 0.9.11", "rustfs-io-metrics", "thiserror 2.0.18", "tokio", @@ -11187,7 +11189,7 @@ checksum = "7abf34aa716da5d5b4c496936d042ea282ab392092cd68a72ef6a8863ff8c96a" dependencies = [ "base64ct", "bytes", - "crypto-bigint 0.7.4", + "crypto-bigint 0.7.5", "ctutils", "digest 0.11.3", "pem-rfc7468 1.0.0", @@ -11374,7 +11376,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "332615d90111d8eeaf86a84dc9bbe9f65d0d8c5cf11b4caccedc37754eb0dcfd" dependencies = [ "debugid", - "memmap2 0.9.10", + "memmap2 0.9.11", "stable_deref_trait", "uuid", ] @@ -11523,7 +11525,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ "fastrand", - "getrandom 0.4.3", + "getrandom 0.3.4", "once_cell", "rustix 1.1.4", "windows-sys 0.61.2", diff --git a/Cargo.toml b/Cargo.toml index 2230698d5..63527b8d3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -243,7 +243,7 @@ lazy_static = "1.5.0" libc = "0.2.186" libsystemd = "0.7.2" local-ip-address = "0.6.13" -memmap2 = "0.9.10" +memmap2 = "0.9.11" lz4 = "1.28.1" matchit = "0.9.2" md-5 = "0.11.0" diff --git a/crates/concurrency/Cargo.toml b/crates/concurrency/Cargo.toml index 2acdc6f75..13bdf1392 100644 --- a/crates/concurrency/Cargo.toml +++ b/crates/concurrency/Cargo.toml @@ -14,6 +14,7 @@ categories = ["concurrency", "filesystem"] # Internal crates rustfs-io-core = { workspace = true } rustfs-io-metrics = { workspace = true } +serde = { workspace = true } # Async runtime tokio = { workspace = true, features = ["sync", "time", "rt"] } @@ -26,6 +27,7 @@ thiserror = { workspace = true } tracing = { workspace = true } [dev-dependencies] +serde_json = { workspace = true } tokio = { workspace = true, features = ["test-util","macros","rt-multi-thread"] } [features] diff --git a/crates/concurrency/src/workload.rs b/crates/concurrency/src/workload.rs index 064e2ede0..b12b66096 100644 --- a/crates/concurrency/src/workload.rs +++ b/crates/concurrency/src/workload.rs @@ -14,8 +14,11 @@ //! Runtime workload admission contract types. +use serde::{Deserialize, Serialize}; + /// Stable workload classes used by future runtime admission snapshots. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] pub enum WorkloadClass { /// Foreground read requests. ForegroundRead, @@ -62,7 +65,8 @@ impl std::fmt::Display for WorkloadClass { } /// Read-only admission state for a workload class. -#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Hash)] +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] pub enum AdmissionState { /// Admission is open for this workload class. Open, @@ -85,7 +89,8 @@ impl AdmissionState { } /// Read-only admission snapshot for one workload class. -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] pub struct WorkloadAdmissionSnapshot { /// Workload class described by this snapshot. pub class: WorkloadClass, @@ -130,7 +135,8 @@ impl WorkloadAdmissionSnapshot { } /// Read-only admission registry snapshot. -#[derive(Debug, Clone, PartialEq, Eq, Default)] +#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] pub struct WorkloadAdmissionRegistrySnapshot { entries: Vec, } @@ -175,6 +181,7 @@ pub trait WorkloadAdmissionSnapshotProvider { #[cfg(test)] mod tests { use super::*; + use serde_json::json; #[test] fn workload_contract_covers_required_classes() { @@ -249,4 +256,39 @@ mod tests { Some(AdmissionState::Open) ); } + + #[test] + fn workload_admission_snapshot_round_trips_with_snake_case_enums() { + let snapshot = WorkloadAdmissionSnapshot::new(WorkloadClass::ForegroundWrite, AdmissionState::Disabled) + .with_counts(Some(0), Some(0), Some(0)) + .with_reason("foreground write admission not exposed"); + let encoded = serde_json::to_value(&snapshot).expect("serialize workload admission snapshot"); + + assert_eq!(encoded["class"], json!("foreground_write")); + assert_eq!(encoded["state"], json!("disabled")); + + let decoded: WorkloadAdmissionSnapshot = + serde_json::from_value(encoded).expect("deserialize workload admission snapshot"); + assert_eq!(decoded, snapshot); + } + + #[test] + fn workload_admission_registry_rejects_unknown_fields() { + let raw = json!({ + "entries": [{ + "class": "foreground_read", + "state": "open", + "active": 1, + "queued": 0, + "limit": 8, + "reason": null, + "unexpected": true + }] + }); + + let err = serde_json::from_value::(raw) + .expect_err("unknown workload admission fields must be rejected"); + + assert!(err.to_string().contains("unexpected")); + } } diff --git a/crates/ecstore/src/cluster/mod.rs b/crates/ecstore/src/cluster/mod.rs index 02d7ba6be..91e82e16c 100644 --- a/crates/ecstore/src/cluster/mod.rs +++ b/crates/ecstore/src/cluster/mod.rs @@ -387,6 +387,7 @@ fn endpoint_labels(endpoint: &Endpoint) -> TopologyLabels { additional.insert(LOCAL_ENDPOINT_LABEL.to_owned(), endpoint.is_local.to_string()); TopologyLabels { + node: Some(endpoint_node_id(endpoint)), additional, ..TopologyLabels::default() } @@ -453,6 +454,7 @@ mod tests { .map(String::as_str), Some("path") ); + assert_eq!(snapshot.pools[0].sets[0].disks[0].labels.node.as_deref(), Some(LOCAL_NODE_ID)); let encoded = serde_json::to_string(&snapshot).expect("serialize topology snapshot"); assert!(!encoded.contains("/tmp/rustfs-cluster-control-plane")); @@ -464,6 +466,7 @@ mod tests { let snapshot = topology_snapshot_from_endpoint_pools(&endpoint_pools); assert_eq!(snapshot.pools[0].sets[0].disks[0].disk_id.as_deref(), Some("node1.example:9000")); + assert_eq!(snapshot.pools[0].sets[0].disks[0].labels.node.as_deref(), Some("node1.example:9000")); assert_eq!( snapshot.pools[0].sets[0].disks[0] .labels diff --git a/deny.toml b/deny.toml index aee901ab8..554d76272 100644 --- a/deny.toml +++ b/deny.toml @@ -19,10 +19,6 @@ no-default-features = false version = 2 yanked = "deny" ignore = [ - # `instant 0.1.13` — unmaintained. No direct dependency; pulled in - # transitively. Tracked for upgrade as part of broader dep refresh. - # owner: rustfs-maintainers review: 2026-07 - { id = "RUSTSEC-2024-0384", reason = "instant unmaintained; transitive only; tracked for upgrade" }, # `paste 1.0.15` — unmaintained. No direct dependency. # owner: rustfs-maintainers review: 2026-07 { id = "RUSTSEC-2024-0436", reason = "paste unmaintained; transitive only; tracked for upgrade" }, @@ -33,10 +29,6 @@ ignore = [ # in-process RSA decryption oracles are removed. # owner: rustfs-maintainers review: 2026-07 { id = "RUSTSEC-2023-0071", reason = "rsa Marvin timing sidechannel; no fixed upstream version; tracked separately" }, - # `proc-macro-error2 2.0.1` — unmaintained. No direct dependency; pulled - # transitively through `mysql_async` via `mysql-common-derive`. - # owner: rustfs-maintainers review: 2026-07 - { id = "RUSTSEC-2026-0173", reason = "proc-macro-error2 unmaintained; transitive only via mysql_async; tracked for dependency refresh" }, ] [sources] diff --git a/rustfs/src/admin/handlers/extensions.rs b/rustfs/src/admin/handlers/extensions.rs index 86431cc98..d6a41c329 100644 --- a/rustfs/src/admin/handlers/extensions.rs +++ b/rustfs/src/admin/handlers/extensions.rs @@ -14,7 +14,7 @@ use crate::admin::{ auth::validate_admin_request, - handlers::plugins_instances, + handlers::{plugins_instances, system}, plugin_contract::{ PluginContractDomain, PluginInstanceDiagnosticCode, PluginInstanceDiagnosticCount, PluginInstanceEntry, PluginInstanceSource, PluginOperationalStateContract, @@ -31,6 +31,7 @@ use rustfs_extension_schema::{ OPS_PROFILER_CAPABILITY, OpsDiagnosticsContract, OpsProfilerContract, }; use rustfs_policy::policy::action::{Action, AdminAction}; +use rustfs_storage_api::CapabilityStatus; use rustfs_targets::{ OpsDiagnosticsRegistry, OpsProfilerRegistry, TargetPluginExternalFlowGate, TargetPluginExternalFlowGateStatus, builtin_extension_schemas, builtin_ops_diagnostics_contract, builtin_ops_diagnostics_extension_schema, @@ -80,6 +81,10 @@ where pub runtime: ExtensionRuntimeContract, pub disabled_by_default: bool, pub startup_fatal: bool, + #[serde(skip_serializing_if = "Option::is_none")] + pub runtime_capability_path: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub runtime_capability_summary: Option, pub contract: T, } @@ -112,20 +117,23 @@ pub(crate) struct ExtensionInstancesResponse { pub next_marker: Option, } -fn build_extension_catalog_response() -> ExtensionCatalogResponse { +async fn build_extension_catalog_response() -> Result { let mut extensions = builtin_extension_schemas(); let example = example_external_webhook_plugin(); extensions.push(target_marketplace_extension_schema(&example.manifest)); extensions.sort_by(|a, b| a.extension_id.cmp(&b.extension_id)); + let runtime_capabilities = system::build_runtime_capabilities_response().await?; - ExtensionCatalogResponse { + Ok(ExtensionCatalogResponse { extensions, - runtime_capabilities: build_extension_runtime_capabilities_response(), + runtime_capabilities: build_extension_runtime_capabilities_response(&runtime_capabilities), external_plugin_flow: TargetPluginExternalFlowGate::default().status(), - } + }) } -fn build_extension_runtime_capabilities_response() -> ExtensionRuntimeCapabilitiesResponse { +fn build_extension_runtime_capabilities_response( + runtime_capabilities: &system::RuntimeCapabilitiesResponse, +) -> ExtensionRuntimeCapabilitiesResponse { let ops_diagnostics_schema = builtin_ops_diagnostics_extension_schema(); let ops_diagnostics_contract = builtin_ops_diagnostics_contract(); let mut ops_diagnostics_registry = OpsDiagnosticsRegistry::new(); @@ -143,6 +151,7 @@ fn build_extension_runtime_capabilities_response() -> ExtensionRuntimeCapabiliti .register_schema(&ops_profiler_schema, &ops_profiler_contract) .is_ok() ); + let runtime_capability_path = Some(format!("{}{}", ADMIN_PREFIX, system::RUNTIME_CAPABILITIES_ROUTE_SUFFIX)); ExtensionRuntimeCapabilitiesResponse { ops_diagnostics: ExtensionRuntimeCapabilityResponse { @@ -151,6 +160,8 @@ fn build_extension_runtime_capabilities_response() -> ExtensionRuntimeCapabiliti runtime: ops_diagnostics_schema.runtime, disabled_by_default: ops_diagnostics_schema.disabled_by_default, startup_fatal: false, + runtime_capability_path: runtime_capability_path.clone(), + runtime_capability_summary: Some(runtime_capabilities.summary.observability.clone()), contract: ops_diagnostics_contract, }, ops_profiler: ExtensionRuntimeCapabilityResponse { @@ -159,6 +170,8 @@ fn build_extension_runtime_capabilities_response() -> ExtensionRuntimeCapabiliti runtime: ops_profiler_schema.runtime, disabled_by_default: ops_profiler_schema.disabled_by_default, startup_fatal: false, + runtime_capability_path, + runtime_capability_summary: Some(runtime_capabilities.summary.userspace_profiling.clone()), contract: ops_profiler_contract, }, } @@ -240,7 +253,10 @@ pub struct GetExtensionCatalogHandler {} impl Operation for GetExtensionCatalogHandler { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { authorize_extension_catalog_request(&req).await?; - build_json_response(StatusCode::OK, &build_extension_catalog_response(), req.headers.get("x-request-id")) + let response = build_extension_catalog_response() + .await + .map_err(|err| s3_error!(InternalError, "failed to build extension catalog response: {}", err))?; + build_json_response(StatusCode::OK, &response, req.headers.get("x-request-id")) } } @@ -267,10 +283,12 @@ impl Operation for ListExtensionInstancesHandler { #[cfg(test)] mod tests { use super::{build_extension_catalog_response, map_extension_instance}; + use crate::admin::handlers::system::{self, RUNTIME_CAPABILITIES_ROUTE_SUFFIX}; use crate::admin::plugin_contract::{ PluginContractDomain, PluginEnableState, PluginInstallState, PluginInstanceEntry, PluginInstanceSource, PluginOperationalRuntimeState, PluginOperationalStateContract, }; + use crate::server::ADMIN_PREFIX; use rustfs_extension_schema::{ ExtensionKind, ExtensionRuntimeBoundary, OPS_DIAGNOSTICS_CAPABILITY, OPS_PROFILER_CAPABILITY, validate_extension_schemas, validate_ops_diagnostics_contract, validate_ops_profiler_contract, @@ -314,9 +332,11 @@ mod tests { ); } - #[test] - fn extension_catalog_exposes_valid_extension_schemas() { - let response = build_extension_catalog_response(); + #[tokio::test] + async fn extension_catalog_exposes_valid_extension_schemas() { + let response = build_extension_catalog_response() + .await + .expect("extension catalog response should build"); let webhook = response .extensions .iter() @@ -373,9 +393,14 @@ mod tests { assert!(validate_extension_schemas(&response.extensions).is_ok()); } - #[test] - fn extension_catalog_exposes_read_only_runtime_capability_snapshots() { - let response = build_extension_catalog_response(); + #[tokio::test] + async fn extension_catalog_exposes_read_only_runtime_capability_snapshots() { + let runtime_capabilities = system::build_runtime_capabilities_response() + .await + .expect("runtime capabilities response should build"); + let response = build_extension_catalog_response() + .await + .expect("extension catalog response should build"); assert_eq!(response.runtime_capabilities.ops_diagnostics.extension_id, "builtin:ops-diagnostics"); assert_eq!( @@ -388,6 +413,22 @@ mod tests { ); assert!(!response.runtime_capabilities.ops_diagnostics.disabled_by_default); assert!(!response.runtime_capabilities.ops_diagnostics.startup_fatal); + assert_eq!( + response + .runtime_capabilities + .ops_diagnostics + .runtime_capability_path + .as_deref(), + Some(format!("{}{}", ADMIN_PREFIX, RUNTIME_CAPABILITIES_ROUTE_SUFFIX).as_str()) + ); + assert_eq!( + response + .runtime_capabilities + .ops_diagnostics + .runtime_capability_summary + .as_ref(), + Some(&runtime_capabilities.summary.observability) + ); assert!(response.runtime_capabilities.ops_diagnostics.contract.requires_admin_action); assert!(!response.runtime_capabilities.ops_diagnostics.contract.mutates_object_data); assert!(validate_ops_diagnostics_contract(&response.runtime_capabilities.ops_diagnostics.contract).is_ok()); @@ -400,6 +441,14 @@ mod tests { ); assert!(!response.runtime_capabilities.ops_profiler.disabled_by_default); assert!(!response.runtime_capabilities.ops_profiler.startup_fatal); + assert_eq!( + response.runtime_capabilities.ops_profiler.runtime_capability_path.as_deref(), + Some(format!("{}{}", ADMIN_PREFIX, RUNTIME_CAPABILITIES_ROUTE_SUFFIX).as_str()) + ); + assert_eq!( + response.runtime_capabilities.ops_profiler.runtime_capability_summary.as_ref(), + Some(&runtime_capabilities.summary.userspace_profiling) + ); assert!(validate_ops_profiler_contract(&response.runtime_capabilities.ops_profiler.contract).is_ok()); } diff --git a/rustfs/src/admin/handlers/system.rs b/rustfs/src/admin/handlers/system.rs index d3ce440ce..765909cb6 100644 --- a/rustfs/src/admin/handlers/system.rs +++ b/rustfs/src/admin/handlers/system.rs @@ -16,14 +16,22 @@ use super::metrics; use crate::admin::auth::validate_admin_request; use crate::admin::router::{AdminOperation, Operation, S3Router}; use crate::app::admin_usecase::{DefaultAdminUsecase, QueryServerInfoRequest}; +use crate::app::context::resolve_endpoints_handle; use crate::auth::{check_key_valid, get_session_token}; +use crate::runtime_capabilities::{EndpointTopologySnapshotProvider, RustFsObservabilitySnapshotProvider}; use crate::server::{ADMIN_PREFIX, RemoteAddr}; +use crate::workload_admission::workload_admission_registry_snapshot; use http::{HeaderMap, HeaderValue}; use hyper::{Method, StatusCode}; use matchit::Params; +use rustfs_concurrency::WorkloadAdmissionRegistrySnapshot; use rustfs_policy::policy::action::{Action, AdminAction, S3Action}; +use rustfs_storage_api::{ + CapabilityState, CapabilityStatus, ObservabilitySnapshotProvider, TopologySnapshot, TopologySnapshotProvider, +}; use s3s::header::CONTENT_TYPE; use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error}; +use serde::Serialize; use tracing::{error, info, warn}; const LOG_COMPONENT_ADMIN_API: &str = "admin_api"; @@ -31,6 +39,10 @@ const LOG_SUBSYSTEM_SYSTEM_ADMIN: &str = "system_admin"; const EVENT_ADMIN_REQUEST_REJECTED: &str = "admin_request_rejected"; const EVENT_ADMIN_REQUEST_FAILED: &str = "admin_request_failed"; const EVENT_ADMIN_RESPONSE_EMITTED: &str = "admin_response_emitted"; +const OBSERVABILITY_SUMMARY_RESOLVED: &str = "observability summary resolved from provider snapshots"; +const TOPOLOGY_SUMMARY_RESOLVED: &str = "topology summary resolved from capability snapshot"; +const TOPOLOGY_SNAPSHOT_NOT_AVAILABLE: &str = "endpoint topology is not available before storage endpoint pools initialize"; +pub(crate) const RUNTIME_CAPABILITIES_ROUTE_SUFFIX: &str = "/v4/runtime/capabilities"; macro_rules! log_system_request_rejected { ($operation:expr, $reason:expr) => { @@ -117,6 +129,12 @@ pub fn register_system_route(r: &mut S3Router) -> std::io::Resul AdminOperation(&metrics::MetricsHandler {}), )?; + r.insert( + Method::GET, + format!("{}{}", ADMIN_PREFIX, RUNTIME_CAPABILITIES_ROUTE_SUFFIX).as_str(), + AdminOperation(&RuntimeCapabilitiesHandler {}), + )?; + Ok(()) } pub struct ServiceHandle {} @@ -225,6 +243,176 @@ impl Operation for StorageInfoHandler { pub struct DataUsageInfoHandler {} +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub struct RuntimeCapabilitiesSummary { + pub observability: CapabilityStatus, + pub userspace_profiling: CapabilityStatus, + pub memory_sampling: CapabilityStatus, + pub platform: CapabilityStatus, + pub topology: CapabilityStatus, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub struct RuntimeCapabilitiesResponse { + pub summary: RuntimeCapabilitiesSummary, + pub observability: rustfs_storage_api::ObservabilitySnapshot, + pub workload_admission: WorkloadAdmissionRegistrySnapshot, + pub topology: Option, + pub topology_status: CapabilityStatus, +} + +pub struct RuntimeCapabilitiesHandler {} + +pub(crate) async fn build_runtime_capabilities_response() +-> Result { + let observability_provider = RustFsObservabilitySnapshotProvider; + let observability = observability_provider.observability_snapshot().await?; + let workload_admission = workload_admission_registry_snapshot(); + + let (topology, topology_status) = if let Some(endpoint_pools) = resolve_endpoints_handle() { + let topology_provider = EndpointTopologySnapshotProvider::new(endpoint_pools); + let topology = topology_provider.topology_snapshot().await?; + (Some(topology), CapabilityStatus::supported()) + } else { + (None, CapabilityStatus::unknown().with_reason(TOPOLOGY_SNAPSHOT_NOT_AVAILABLE)) + }; + let summary = build_runtime_capabilities_summary(&observability, topology.as_ref(), &topology_status); + + Ok(RuntimeCapabilitiesResponse { + summary, + observability, + workload_admission, + topology, + topology_status, + }) +} + +fn build_runtime_capabilities_summary( + observability: &rustfs_storage_api::ObservabilitySnapshot, + topology: Option<&TopologySnapshot>, + topology_status: &CapabilityStatus, +) -> RuntimeCapabilitiesSummary { + let userspace_profiling = summarize_named_capability_statuses( + [ + ("cpu", &observability.userspace_profiling.cpu), + ("memory", &observability.userspace_profiling.memory), + ("continuous_cpu", &observability.userspace_profiling.continuous_cpu), + ("periodic_cpu", &observability.userspace_profiling.periodic_cpu), + ], + "userspace profiling", + ); + let memory_sampling = summarize_named_capability_statuses( + [ + ("process", &observability.memory_sampling.process), + ("system", &observability.memory_sampling.system), + ("cgroup", &observability.memory_sampling.cgroup), + ], + "memory sampling", + ); + let platform = summarize_named_capability_statuses( + [ + ("allocator", &observability.platform.allocator), + ("ebpf", &observability.platform.ebpf), + ("numa", &observability.platform.numa), + ], + "platform support", + ); + let observability_status = if [userspace_profiling.state, memory_sampling.state, platform.state] + .into_iter() + .any(|state| state == CapabilityState::Unknown) + { + CapabilityStatus::unknown().with_reason(OBSERVABILITY_SUMMARY_RESOLVED) + } else { + CapabilityStatus::supported().with_reason(OBSERVABILITY_SUMMARY_RESOLVED) + }; + let topology_summary = topology.map_or_else( + || topology_status.clone(), + |topology| { + summarize_named_capability_statuses( + [ + ("profiling", &topology.capabilities.profiling), + ("numa", &topology.capabilities.numa), + ("failure_domain_labels", &topology.capabilities.failure_domain_labels), + ("media_labels", &topology.capabilities.media_labels), + ], + "topology capability", + ) + .with_reason(TOPOLOGY_SUMMARY_RESOLVED) + }, + ); + + RuntimeCapabilitiesSummary { + observability: observability_status, + userspace_profiling, + memory_sampling, + platform, + topology: topology_summary, + } +} + +fn summarize_named_capability_statuses( + statuses: [(&'static str, &CapabilityStatus); N], + subject: &'static str, +) -> CapabilityStatus { + let unknown = statuses + .iter() + .filter_map(|(name, status)| (status.state == CapabilityState::Unknown).then_some(*name)) + .collect::>(); + if !unknown.is_empty() { + return CapabilityStatus::unknown().with_reason(format!("{subject} unresolved fields: {}", unknown.join(", "))); + } + + let supported_or_disabled = statuses + .iter() + .any(|(_, status)| matches!(status.state, CapabilityState::Supported | CapabilityState::Disabled)); + if supported_or_disabled { + return CapabilityStatus::supported(); + } + + let unsupported = statuses.iter().map(|(name, _)| *name).collect::>(); + CapabilityStatus::unsupported().with_reason(format!("{subject} unsupported fields: {}", unsupported.join(", "))) +} + +#[async_trait::async_trait] +impl Operation for RuntimeCapabilitiesHandler { + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + let Some(input_cred) = req.credentials else { + log_system_request_rejected!("runtime_capabilities", "missing_credentials"); + return Err(s3_error!(InvalidRequest, "get cred failed")); + }; + + let (cred, owner) = + check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?; + + let remote_addr = req.extensions.get::>().and_then(|opt| opt.map(|a| a.0)); + validate_admin_request( + &req.headers, + &cred, + owner, + false, + vec![Action::AdminAction(AdminAction::ServerInfoAdminAction)], + remote_addr, + ) + .await?; + + let response = build_runtime_capabilities_response().await.map_err(|err| { + log_system_request_failed!("runtime_capabilities", "build_runtime_capabilities_failed", err); + S3Error::with_message(S3ErrorCode::InternalError, "failed to build runtime capabilities snapshot") + })?; + + let data = serde_json::to_vec(&response).map_err(|e| { + log_system_request_failed!("runtime_capabilities", "serialize_runtime_capabilities_failed", e); + S3Error::with_message(S3ErrorCode::InternalError, "failed to serialize runtime capabilities snapshot") + })?; + + let mut header = HeaderMap::new(); + header.insert(CONTENT_TYPE, HeaderValue::from_static("application/json")); + log_system_response_emitted!("runtime_capabilities"); + + Ok(S3Response::with_headers((StatusCode::OK, Body::from(data)), header)) + } +} + #[async_trait::async_trait] impl Operation for DataUsageInfoHandler { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { @@ -265,3 +453,149 @@ impl Operation for DataUsageInfoHandler { Ok(S3Response::with_headers((StatusCode::OK, Body::from(data)), header)) } } + +#[cfg(test)] +mod tests { + use super::{ + OBSERVABILITY_SUMMARY_RESOLVED, TOPOLOGY_SNAPSHOT_NOT_AVAILABLE, TOPOLOGY_SUMMARY_RESOLVED, + build_runtime_capabilities_response, build_runtime_capabilities_summary, + }; + use rustfs_concurrency::{AdmissionState, WorkloadClass}; + use rustfs_storage_api::{ + CapabilityState, CapabilityStatus, MemorySamplingState, ObservabilitySnapshot, PlatformSupport, TopologyCapabilities, + TopologySnapshot, UserspaceProfilingCapability, + }; + + #[tokio::test] + async fn runtime_capabilities_response_reports_missing_topology_before_storage_init() { + let response = build_runtime_capabilities_response() + .await + .expect("runtime capabilities response should build"); + + assert_eq!(response.topology, None); + assert_eq!(response.topology_status.state, CapabilityState::Unknown); + assert_eq!(response.topology_status.reason.as_deref(), Some(TOPOLOGY_SNAPSHOT_NOT_AVAILABLE)); + assert_eq!(response.summary.topology.state, CapabilityState::Unknown); + assert_eq!(response.summary.topology.reason.as_deref(), Some(TOPOLOGY_SNAPSHOT_NOT_AVAILABLE)); + assert_eq!(response.observability.platform.os.as_deref(), Some(std::env::consts::OS)); + assert_eq!( + response + .workload_admission + .get(WorkloadClass::ForegroundRead) + .map(|snapshot| snapshot.state), + Some(AdmissionState::Open) + ); + assert_eq!( + response + .workload_admission + .get(WorkloadClass::ForegroundWrite) + .map(|snapshot| snapshot.state), + Some(AdmissionState::Disabled) + ); + assert_eq!(response.workload_admission.entries().len(), WorkloadClass::REQUIRED.len()); + } + + #[test] + fn runtime_capabilities_summary_marks_observability_supported_when_groups_are_resolved() { + let observability = ObservabilitySnapshot { + runtime_telemetry: CapabilityStatus::supported(), + userspace_profiling: UserspaceProfilingCapability { + cpu: CapabilityStatus::supported(), + memory: CapabilityStatus::unsupported().with_reason("platform"), + continuous_cpu: CapabilityStatus::supported(), + periodic_cpu: CapabilityStatus::disabled(), + }, + memory_sampling: MemorySamplingState { + process: CapabilityStatus::supported(), + system: CapabilityStatus::supported(), + cgroup: CapabilityStatus::unsupported().with_reason("platform"), + }, + platform: PlatformSupport { + target_triple: Some("x86_64-unknown-linux-gnu".to_owned()), + os: Some("linux".to_owned()), + arch: Some("x86_64".to_owned()), + allocator: CapabilityStatus::supported(), + ebpf: CapabilityStatus::unsupported().with_reason("platform"), + numa: CapabilityStatus::unsupported().with_reason("platform"), + }, + }; + let topology = TopologySnapshot { + pools: Vec::new(), + capabilities: TopologyCapabilities { + profiling: CapabilityStatus::supported(), + numa: CapabilityStatus::unsupported().with_reason("platform"), + failure_domain_labels: CapabilityStatus::supported(), + media_labels: CapabilityStatus::unsupported().with_reason("not reported"), + }, + }; + + let summary = build_runtime_capabilities_summary(&observability, Some(&topology), &CapabilityStatus::supported()); + + assert_eq!(summary.observability.state, CapabilityState::Supported); + assert_eq!(summary.observability.reason.as_deref(), Some(OBSERVABILITY_SUMMARY_RESOLVED)); + assert_eq!(summary.userspace_profiling.state, CapabilityState::Supported); + assert_eq!(summary.memory_sampling.state, CapabilityState::Supported); + assert_eq!(summary.platform.state, CapabilityState::Supported); + assert_eq!(summary.topology.state, CapabilityState::Supported); + assert_eq!(summary.topology.reason.as_deref(), Some(TOPOLOGY_SUMMARY_RESOLVED)); + } + + #[test] + fn runtime_capabilities_summary_marks_unknown_groups_when_snapshot_is_not_fully_wired() { + let observability = ObservabilitySnapshot { + runtime_telemetry: CapabilityStatus::supported(), + userspace_profiling: UserspaceProfilingCapability { + cpu: CapabilityStatus::unknown().with_reason("not wired"), + memory: CapabilityStatus::unsupported().with_reason("platform"), + continuous_cpu: CapabilityStatus::supported(), + periodic_cpu: CapabilityStatus::disabled(), + }, + memory_sampling: MemorySamplingState { + process: CapabilityStatus::supported(), + system: CapabilityStatus::supported(), + cgroup: CapabilityStatus::unknown().with_reason("not wired"), + }, + platform: PlatformSupport { + target_triple: Some("x86_64-unknown-linux-gnu".to_owned()), + os: Some("linux".to_owned()), + arch: Some("x86_64".to_owned()), + allocator: CapabilityStatus::supported(), + ebpf: CapabilityStatus::unknown().with_reason("not wired"), + numa: CapabilityStatus::unknown().with_reason("not wired"), + }, + }; + let topology = TopologySnapshot { + pools: Vec::new(), + capabilities: TopologyCapabilities { + profiling: CapabilityStatus::supported(), + numa: CapabilityStatus::unknown().with_reason("not wired"), + failure_domain_labels: CapabilityStatus::unknown().with_reason("not reported"), + media_labels: CapabilityStatus::unknown().with_reason("not reported"), + }, + }; + + let summary = build_runtime_capabilities_summary(&observability, Some(&topology), &CapabilityStatus::supported()); + + assert_eq!(summary.observability.state, CapabilityState::Unknown); + assert_eq!(summary.userspace_profiling.state, CapabilityState::Unknown); + assert_eq!(summary.memory_sampling.state, CapabilityState::Unknown); + assert_eq!(summary.platform.state, CapabilityState::Unknown); + assert_eq!(summary.topology.state, CapabilityState::Unknown); + assert!( + summary + .userspace_profiling + .reason + .as_deref() + .unwrap_or_default() + .contains("cpu") + ); + assert!( + summary + .topology + .reason + .as_deref() + .unwrap_or_default() + .contains(TOPOLOGY_SUMMARY_RESOLVED) + ); + } +} diff --git a/rustfs/src/admin/route_policy.rs b/rustfs/src/admin/route_policy.rs index aa81d2e41..772bddc20 100644 --- a/rustfs/src/admin/route_policy.rs +++ b/rustfs/src/admin/route_policy.rs @@ -350,6 +350,12 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[ GET_BUCKET_TARGET, RouteRiskLevel::Sensitive, ), + admin( + HttpMethod::Get, + "/rustfs/admin/v4/runtime/capabilities", + SERVER_INFO, + RouteRiskLevel::Sensitive, + ), admin( HttpMethod::Get, "/rustfs/admin/v4/plugins/catalog", diff --git a/rustfs/src/admin/route_registration_test.rs b/rustfs/src/admin/route_registration_test.rs index 27e118c5b..874d6ff66 100644 --- a/rustfs/src/admin/route_registration_test.rs +++ b/rustfs/src/admin/route_registration_test.rs @@ -226,6 +226,7 @@ fn expected_admin_route_matrix() -> Vec { admin_route(Method::PUT, "/v3/module-switches"), admin_route(Method::GET, "/v4/extensions/catalog"), admin_route(Method::GET, "/v4/extensions/instances"), + admin_route(Method::GET, "/v4/runtime/capabilities"), admin_route(Method::POST, "/v3/object-zip-downloads"), admin_route_sample( Method::GET, @@ -724,6 +725,7 @@ fn test_register_routes_cover_representative_admin_paths() { assert_route(&router, Method::PUT, &admin_path("/v3/module-switches")); assert_route(&router, Method::GET, &admin_path("/v4/extensions/catalog")); assert_route(&router, Method::GET, &admin_path("/v4/extensions/instances")); + assert_route(&router, Method::GET, &admin_path("/v4/runtime/capabilities")); assert_route(&router, Method::POST, &admin_path("/v3/object-zip-downloads")); assert_route(&router, Method::GET, &admin_path("/v4/plugins/catalog")); assert_route(&router, Method::GET, &admin_path("/v4/plugins/instances")); diff --git a/rustfs/src/runtime_capabilities.rs b/rustfs/src/runtime_capabilities.rs index fdecdc496..abf7fafe4 100644 --- a/rustfs/src/runtime_capabilities.rs +++ b/rustfs/src/runtime_capabilities.rs @@ -20,9 +20,11 @@ use rustfs_storage_api::{ }; const NOT_WIRED_INTO_RUNTIME: &str = "not wired into runtime"; +const EBPF_LINUX_ONLY: &str = "eBPF support is only available on linux targets"; const STORAGE_MEDIA_NOT_REPORTED: &str = "storage media not reported by endpoints"; const FAILURE_DOMAIN_NOT_REPORTED: &str = "failure domain labels not reported by endpoints"; const NUMA_NOT_WIRED: &str = "NUMA topology not wired into runtime"; +const NUMA_LINUX_ONLY: &str = "NUMA topology reporting currently targets linux runtimes"; #[derive(Debug, Default, Clone, Copy)] pub struct RustFsObservabilitySnapshotProvider; @@ -49,13 +51,13 @@ pub fn runtime_observability_snapshot() -> ObservabilitySnapshot { cgroup: cgroup_memory_status(), }, platform: PlatformSupport { - target_triple: option_env!("TARGET").map(str::to_owned), + target_triple: Some(compiled_target_triple()), os: Some(std::env::consts::OS.to_owned()), arch: Some(std::env::consts::ARCH.to_owned()), allocator: CapabilityStatus::supported() .with_reason(format!("backend={}", crate::allocator_reclaim::allocator_backend())), - ebpf: CapabilityStatus::unknown().with_reason(NOT_WIRED_INTO_RUNTIME), - numa: CapabilityStatus::unsupported().with_reason(NUMA_NOT_WIRED), + ebpf: ebpf_status(), + numa: numa_status(), }, } } @@ -104,6 +106,42 @@ fn runtime_telemetry_status() -> CapabilityStatus { } } +fn compiled_target_triple() -> String { + let arch = std::env::consts::ARCH; + let vendor = target_vendor_name(); + let os = std::env::consts::OS; + let env = target_env_name(); + + match env { + Some(env) => format!("{arch}-{vendor}-{os}-{env}"), + None => format!("{arch}-{vendor}-{os}"), + } +} + +fn target_vendor_name() -> &'static str { + if cfg!(target_vendor = "apple") { + "apple" + } else if cfg!(target_vendor = "pc") { + "pc" + } else { + "unknown" + } +} + +fn target_env_name() -> Option<&'static str> { + if cfg!(target_env = "gnu") { + Some("gnu") + } else if cfg!(target_env = "musl") { + Some("musl") + } else if cfg!(target_env = "msvc") { + Some("msvc") + } else if cfg!(target_env = "sgx") { + Some("sgx") + } else { + None + } +} + fn cpu_profiling_status() -> CapabilityStatus { if cfg!(any(target_os = "linux", target_os = "macos")) { CapabilityStatus::supported() @@ -112,6 +150,22 @@ fn cpu_profiling_status() -> CapabilityStatus { } } +fn ebpf_status() -> CapabilityStatus { + if cfg!(target_os = "linux") { + CapabilityStatus::unknown().with_reason(NOT_WIRED_INTO_RUNTIME) + } else { + CapabilityStatus::unsupported().with_reason(EBPF_LINUX_ONLY) + } +} + +fn numa_status() -> CapabilityStatus { + if cfg!(target_os = "linux") { + CapabilityStatus::unknown().with_reason(NUMA_NOT_WIRED) + } else { + CapabilityStatus::unsupported().with_reason(NUMA_LINUX_ONLY) + } +} + fn memory_profiling_status() -> CapabilityStatus { if cfg!(all(target_os = "linux", target_env = "gnu", target_arch = "x86_64")) { CapabilityStatus::supported().with_reason("jemalloc memory profiling target") @@ -141,6 +195,7 @@ mod tests { assert_eq!(snapshot.platform.os.as_deref(), Some(std::env::consts::OS)); assert_eq!(snapshot.platform.arch.as_deref(), Some(std::env::consts::ARCH)); + assert_eq!(snapshot.platform.target_triple.as_deref(), Some(compiled_target_triple().as_str())); assert!(matches!( snapshot.runtime_telemetry.state, CapabilityState::Supported | CapabilityState::Disabled @@ -154,6 +209,13 @@ mod tests { .unwrap_or_default() .contains("backend=") ); + if cfg!(target_os = "linux") { + assert_eq!(snapshot.platform.ebpf.state, CapabilityState::Unknown); + assert_eq!(snapshot.platform.numa.state, CapabilityState::Unknown); + } else { + assert_eq!(snapshot.platform.ebpf.state, CapabilityState::Unsupported); + assert_eq!(snapshot.platform.numa.state, CapabilityState::Unsupported); + } } #[tokio::test] diff --git a/rustfs/src/workload_admission.rs b/rustfs/src/workload_admission.rs index 21472b8eb..bf4b7dc5f 100644 --- a/rustfs/src/workload_admission.rs +++ b/rustfs/src/workload_admission.rs @@ -22,10 +22,14 @@ use rustfs_concurrency::{ use crate::storage::concurrency::get_concurrency_manager; const BUCKET_METADATA_RUNTIME_NOT_INITIALIZED: &str = "bucket metadata runtime not initialized"; +const FOREGROUND_WRITE_NOT_EXPOSED_BY_PROVIDER: &str = "foreground write admission not yet exposed by RustFS runtime"; const HEAL_MANAGER_NOT_INITIALIZED: &str = "heal manager not initialized"; -const NOT_EXPOSED_BY_PROVIDER: &str = "not exposed by RustFS workload admission provider"; +const REPAIR_QUEUE_BACKLOG_PRESENT: &str = "repair queue has pending work"; const REPLICATION_RUNTIME_NOT_INITIALIZED: &str = "replication runtime not initialized"; +const REPLICATION_QUEUE_BACKLOG_PRESENT: &str = "replication queue has pending work"; const REPLICATION_QUEUE_STATS_UNAVAILABLE: &str = "replication queue stats unavailable"; +const SCANNER_ADMISSION_DISABLED: &str = "scanner admission disabled because max concurrent set scans is zero"; +const SCANNER_ADMISSION_SATURATED: &str = "scanner active work reached configured set-scan limit"; const SCANNER_ACTIVITY_IDLE_OR_NOT_INITIALIZED: &str = "scanner activity idle or not initialized"; const STORAGE_CONCURRENCY_PROVIDER_MISSING_FOREGROUND_READ: &str = "storage concurrency provider did not expose foreground read admission"; @@ -49,18 +53,13 @@ fn storage_concurrency_workload_admission_snapshot() -> WorkloadAdmissionRegistr } fn runtime_owner_workload_admission_registry_snapshot() -> WorkloadAdmissionRegistrySnapshot { - let entries = WorkloadClass::REQUIRED - .iter() - .copied() - .filter_map(|class| match class { - WorkloadClass::ForegroundRead => None, - WorkloadClass::Metadata => Some(metadata_workload_admission_snapshot()), - WorkloadClass::Scanner => Some(scanner_workload_admission_snapshot()), - WorkloadClass::Repair => Some(repair_workload_admission_snapshot()), - WorkloadClass::Replication => Some(replication_workload_admission_snapshot()), - class => Some(WorkloadAdmissionSnapshot::new(class, AdmissionState::Unknown).with_reason(NOT_EXPOSED_BY_PROVIDER)), - }) - .collect(); + let entries = vec![ + foreground_write_workload_admission_snapshot(), + metadata_workload_admission_snapshot(), + scanner_workload_admission_snapshot(), + repair_workload_admission_snapshot(), + replication_workload_admission_snapshot(), + ]; WorkloadAdmissionRegistrySnapshot::new(entries) } @@ -75,6 +74,12 @@ pub fn foreground_read_workload_admission_snapshot() -> WorkloadAdmissionSnapsho }) } +pub fn foreground_write_workload_admission_snapshot() -> WorkloadAdmissionSnapshot { + WorkloadAdmissionSnapshot::new(WorkloadClass::ForegroundWrite, AdmissionState::Disabled) + .with_counts(Some(0), None, Some(0)) + .with_reason(FOREGROUND_WRITE_NOT_EXPOSED_BY_PROVIDER) +} + pub fn metadata_workload_admission_snapshot() -> WorkloadAdmissionSnapshot { metadata_workload_admission_snapshot_from_initialized(bucket_metadata_runtime_initialized()) } @@ -96,11 +101,19 @@ fn metadata_workload_admission_snapshot_from_initialized(runtime_initialized: bo } pub fn scanner_workload_admission_snapshot() -> WorkloadAdmissionSnapshot { - scanner_workload_admission_snapshot_from_activity(rustfs_scanner::current_scanner_activity()) + let runtime_config = rustfs_scanner::scanner_runtime_config_status(); + scanner_workload_admission_snapshot_from_activity( + rustfs_scanner::current_scanner_activity(), + runtime_config.max_concurrent_set_scans.value, + ) } -fn scanner_workload_admission_snapshot_from_activity(active: u64) -> WorkloadAdmissionSnapshot { - let state = if active > 0 { +fn scanner_workload_admission_snapshot_from_activity(active: u64, limit: usize) -> WorkloadAdmissionSnapshot { + let state = if limit == 0 { + AdmissionState::Disabled + } else if usize::try_from(active).ok().is_some_and(|active| active >= limit) { + AdmissionState::Saturated + } else if active > 0 { AdmissionState::Open } else { AdmissionState::Unknown @@ -109,13 +122,14 @@ fn scanner_workload_admission_snapshot_from_activity(active: u64) -> WorkloadAdm let snapshot = WorkloadAdmissionSnapshot::new(WorkloadClass::Scanner, state).with_counts( Some(u64_to_usize_saturated(active)), None, - None, + Some(limit), ); - if state == AdmissionState::Unknown { - snapshot.with_reason(SCANNER_ACTIVITY_IDLE_OR_NOT_INITIALIZED) - } else { - snapshot + match state { + AdmissionState::Disabled => snapshot.with_reason(SCANNER_ADMISSION_DISABLED), + AdmissionState::Saturated => snapshot.with_reason(SCANNER_ADMISSION_SATURATED), + AdmissionState::Unknown => snapshot.with_reason(SCANNER_ACTIVITY_IDLE_OR_NOT_INITIALIZED), + _ => snapshot, } } @@ -132,7 +146,11 @@ fn repair_workload_admission_snapshot_from_counts( active: u64, queued: u64, ) -> WorkloadAdmissionSnapshot { - let state = if manager_initialized || active > 0 || queued > 0 { + let state = if !manager_initialized && active == 0 && queued == 0 { + AdmissionState::Unknown + } else if queued > 0 { + AdmissionState::Throttled + } else if manager_initialized || active > 0 { AdmissionState::Open } else { AdmissionState::Unknown @@ -144,10 +162,10 @@ fn repair_workload_admission_snapshot_from_counts( None, ); - if state == AdmissionState::Unknown { - snapshot.with_reason(HEAL_MANAGER_NOT_INITIALIZED) - } else { - snapshot + match state { + AdmissionState::Unknown => snapshot.with_reason(HEAL_MANAGER_NOT_INITIALIZED), + AdmissionState::Throttled => snapshot.with_reason(REPAIR_QUEUE_BACKLOG_PRESENT), + _ => snapshot, } } @@ -170,7 +188,11 @@ fn replication_workload_admission_snapshot_from_counts( active: Option, queued: Option, ) -> WorkloadAdmissionSnapshot { - let state = if runtime_initialized && queued.is_some() { + let state = if !runtime_initialized { + AdmissionState::Unknown + } else if queued.is_some_and(|queued| queued > 0) { + AdmissionState::Throttled + } else if queued.is_some() { AdmissionState::Open } else { AdmissionState::Unknown @@ -178,12 +200,11 @@ fn replication_workload_admission_snapshot_from_counts( let snapshot = WorkloadAdmissionSnapshot::new(WorkloadClass::Replication, state).with_counts(active, queued, None); - if !runtime_initialized { - snapshot.with_reason(REPLICATION_RUNTIME_NOT_INITIALIZED) - } else if queued.is_none() { - snapshot.with_reason(REPLICATION_QUEUE_STATS_UNAVAILABLE) - } else { - snapshot + match state { + AdmissionState::Unknown if !runtime_initialized => snapshot.with_reason(REPLICATION_RUNTIME_NOT_INITIALIZED), + AdmissionState::Unknown => snapshot.with_reason(REPLICATION_QUEUE_STATS_UNAVAILABLE), + AdmissionState::Throttled => snapshot.with_reason(REPLICATION_QUEUE_BACKLOG_PRESENT), + _ => snapshot, } } @@ -238,38 +259,70 @@ mod tests { assert_eq!(snapshot.reason.as_deref(), Some(BUCKET_METADATA_RUNTIME_NOT_INITIALIZED)); } + #[test] + fn foreground_write_snapshot_reports_explicit_gap_reason() { + let snapshot = foreground_write_workload_admission_snapshot(); + + assert_eq!(snapshot.class, WorkloadClass::ForegroundWrite); + assert_eq!(snapshot.state, AdmissionState::Disabled); + assert_eq!(snapshot.active, Some(0)); + assert_eq!(snapshot.limit, Some(0)); + assert_eq!(snapshot.reason.as_deref(), Some(FOREGROUND_WRITE_NOT_EXPOSED_BY_PROVIDER)); + } + #[test] fn scanner_snapshot_reports_active_work_units() { - let snapshot = scanner_workload_admission_snapshot_from_activity(5); + let snapshot = scanner_workload_admission_snapshot_from_activity(5, 8); assert_eq!(snapshot.class, WorkloadClass::Scanner); assert_eq!(snapshot.state, AdmissionState::Open); assert_eq!(snapshot.active, Some(5)); assert_eq!(snapshot.queued, None); - assert_eq!(snapshot.limit, None); + assert_eq!(snapshot.limit, Some(8)); assert_eq!(snapshot.reason, None); } #[test] fn scanner_snapshot_is_unknown_when_idle_or_uninitialized() { - let snapshot = scanner_workload_admission_snapshot_from_activity(0); + let snapshot = scanner_workload_admission_snapshot_from_activity(0, 8); assert_eq!(snapshot.class, WorkloadClass::Scanner); assert_eq!(snapshot.state, AdmissionState::Unknown); assert_eq!(snapshot.active, Some(0)); + assert_eq!(snapshot.limit, Some(8)); assert_eq!(snapshot.reason.as_deref(), Some(SCANNER_ACTIVITY_IDLE_OR_NOT_INITIALIZED)); } + #[test] + fn scanner_snapshot_reports_disabled_when_set_scan_limit_is_zero() { + let snapshot = scanner_workload_admission_snapshot_from_activity(0, 0); + + assert_eq!(snapshot.class, WorkloadClass::Scanner); + assert_eq!(snapshot.state, AdmissionState::Disabled); + assert_eq!(snapshot.limit, Some(0)); + assert_eq!(snapshot.reason.as_deref(), Some(SCANNER_ADMISSION_DISABLED)); + } + + #[test] + fn scanner_snapshot_reports_saturation_when_active_work_reaches_limit() { + let snapshot = scanner_workload_admission_snapshot_from_activity(4, 4); + + assert_eq!(snapshot.class, WorkloadClass::Scanner); + assert_eq!(snapshot.state, AdmissionState::Saturated); + assert_eq!(snapshot.limit, Some(4)); + assert_eq!(snapshot.reason.as_deref(), Some(SCANNER_ADMISSION_SATURATED)); + } + #[test] fn repair_snapshot_reports_heal_counters() { let snapshot = repair_workload_admission_snapshot_from_counts(true, 2, 3); assert_eq!(snapshot.class, WorkloadClass::Repair); - assert_eq!(snapshot.state, AdmissionState::Open); + assert_eq!(snapshot.state, AdmissionState::Throttled); assert_eq!(snapshot.active, Some(2)); assert_eq!(snapshot.queued, Some(3)); assert_eq!(snapshot.limit, None); - assert_eq!(snapshot.reason, None); + assert_eq!(snapshot.reason.as_deref(), Some(REPAIR_QUEUE_BACKLOG_PRESENT)); } #[test] @@ -288,11 +341,11 @@ mod tests { let snapshot = replication_workload_admission_snapshot_from_counts(true, Some(4), Some(9)); assert_eq!(snapshot.class, WorkloadClass::Replication); - assert_eq!(snapshot.state, AdmissionState::Open); + assert_eq!(snapshot.state, AdmissionState::Throttled); assert_eq!(snapshot.active, Some(4)); assert_eq!(snapshot.queued, Some(9)); assert_eq!(snapshot.limit, None); - assert_eq!(snapshot.reason, None); + assert_eq!(snapshot.reason.as_deref(), Some(REPLICATION_QUEUE_BACKLOG_PRESENT)); } #[test] @@ -350,8 +403,8 @@ mod tests { assert_eq!( registry .get(WorkloadClass::ForegroundWrite) - .and_then(|snapshot| snapshot.reason.as_deref()), - Some(NOT_EXPOSED_BY_PROVIDER) + .map(|snapshot| (snapshot.state, snapshot.reason.as_deref())), + Some((AdmissionState::Disabled, Some(FOREGROUND_WRITE_NOT_EXPOSED_BY_PROVIDER))) ); } } diff --git a/scripts/check_architecture_migration_rules.sh b/scripts/check_architecture_migration_rules.sh index 79ffc64e4..c54e857ad 100755 --- a/scripts/check_architecture_migration_rules.sh +++ b/scripts/check_architecture_migration_rules.sh @@ -448,7 +448,7 @@ require_source_contains \ "RustFS metadata workload admission snapshot helper" require_source_contains \ "rustfs/src/workload_admission.rs" \ - "WorkloadClass::Metadata => Some(metadata_workload_admission_snapshot())" \ + "metadata_workload_admission_snapshot()," \ "RustFS metadata workload admission class mapping" require_source_contains \ "rustfs/src/workload_admission.rs" \ @@ -456,7 +456,7 @@ require_source_contains \ "RustFS scanner workload admission snapshot helper" require_source_contains \ "rustfs/src/workload_admission.rs" \ - "WorkloadClass::Scanner => Some(scanner_workload_admission_snapshot())" \ + "scanner_workload_admission_snapshot()," \ "RustFS scanner workload admission class mapping" require_source_contains \ "rustfs/src/workload_admission.rs" \ @@ -464,7 +464,7 @@ require_source_contains \ "RustFS repair workload admission snapshot helper" require_source_contains \ "rustfs/src/workload_admission.rs" \ - "WorkloadClass::Repair => Some(repair_workload_admission_snapshot())" \ + "repair_workload_admission_snapshot()," \ "RustFS repair workload admission class mapping" require_source_contains \ "rustfs/src/workload_admission.rs" \ @@ -472,7 +472,7 @@ require_source_contains \ "RustFS replication workload admission snapshot helper" require_source_contains \ "rustfs/src/workload_admission.rs" \ - "WorkloadClass::Replication => Some(replication_workload_admission_snapshot())" \ + "replication_workload_admission_snapshot()," \ "RustFS replication workload admission class mapping" ( @@ -1439,10 +1439,18 @@ fi rustfs/src/admin/service/mod.rs \ rustfs/src/app/context.rs \ rustfs/src/storage/rpc/mod.rs || true - rg -n --with-filename --pcre2 '(?/dev/null 2>&1; then + rg -n --with-filename --pcre2 '(?"$RUSTFS_NESTED_SECONDARY_COMPAT_BRIDGE_HITS_FILE"