From 80b1fca02ab5c7500c57e73a0cdd039929e5168e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=AE=89=E6=AD=A3=E8=B6=85?= Date: Fri, 19 Jun 2026 09:50:42 +0800 Subject: [PATCH] feat: wire runtime capability snapshot providers (#3604) --- docs/architecture/migration-progress.md | 58 ++-- .../runtime-capability-contracts.md | 17 ++ rustfs/src/lib.rs | 1 + rustfs/src/runtime_capabilities.rs | 286 ++++++++++++++++++ scripts/check_architecture_migration_rules.sh | 16 + 5 files changed, 359 insertions(+), 19 deletions(-) create mode 100644 rustfs/src/runtime_capabilities.rs diff --git a/docs/architecture/migration-progress.md b/docs/architecture/migration-progress.md index e5156ca95..efbf293ff 100644 --- a/docs/architecture/migration-progress.md +++ b/docs/architecture/migration-progress.md @@ -5,19 +5,18 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block ## Current Context - Issue: [`rustfs/backlog#660`](https://github.com/rustfs/backlog/issues/660) -- Branch: `overtrue/arch-set-local-scheduler-snapshots` -- Baseline: `origin/main` after `rustfs/rustfs#3602` - (`56c3cf50ae93d6705cf9f615e49b66787cb99954`). -- PR type for this branch: `api-extraction` +- Branch: `overtrue/arch-runtime-capability-snapshots` +- Baseline: `origin/main` after `rustfs/rustfs#3603` + (`b106b628c164bf6cdca9e71ff5118711f99d4ab0`). +- PR type for this branch: `consumer-migration` - Runtime behavior changes: none. -- Rust code changes: expose local foreground-read admission snapshots from - RustFS storage concurrency through the `rustfs-concurrency` workload contract. -- CI/script changes: extend migration guard coverage for the storage - `WorkloadAdmissionSnapshotProvider` implementation. -- Docs changes: add - set-local snapshot extraction notes to - [`workload-admission-contracts.md`](workload-admission-contracts.md) and - record the API-055/SCH-001 extraction slice. +- Rust code changes: wire read-only observability and endpoint-topology + snapshot providers in the RustFS runtime. +- CI/script changes: extend migration guard coverage for RustFS runtime + capability provider implementations. +- Docs changes: add RustFS provider notes to + [`runtime-capability-contracts.md`](runtime-capability-contracts.md) and + record the API-056/R-016 provider slice. ## Phase 0 Tasks @@ -157,6 +156,20 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block - Verification: storage concurrency tests, focused RustFS library check, migration guard, formatting, diff hygiene, and three-expert review. +- [x] `API-056/R-016` Wire runtime capability snapshot providers. + - Completed slice: implement `ObservabilitySnapshotProvider` for RustFS + runtime capability state and `TopologySnapshotProvider` for + `EndpointServerPools` topology snapshots. + - Acceptance: observability and endpoint topology snapshots are available + through the storage-api contracts without admin routes, sidecars, ECStore + placement mutation, profiling startup changes, or endpoint behavior changes. + - Must preserve: profiling opt-in behavior, memory and cgroup sampling + behavior, endpoint pool/set/disk assignment, placement, readiness, locks, + and local path privacy. + - Verification: focused runtime capability tests, focused RustFS library + check, migration and layer guards, formatting, diff hygiene, risk scan, and + three-expert review. + - [x] `TEST-PRTYPE-001` Check PR type enum consistency. - Acceptance: `./scripts/check_architecture_migration_rules.sh` parses the allowed PR types from [`crate-boundaries.md`](crate-boundaries.md) and fails @@ -1739,13 +1752,10 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block ## Next PRs -1. `contract`/`consumer-migration`: wire read-only observability and topology - snapshots to implementation owners without changing runtime, profiling, - placement, or admin route behavior. -2. `consumer-migration`: connect additional scanner, repair, replication, and +1. `consumer-migration`: connect additional scanner, repair, replication, and metadata admission owners to the workload registry only after each owner has dedicated preservation coverage. -3. `pure-move`/`consumer-migration`: continue larger cleanup slices with the +2. `pure-move`/`consumer-migration`: continue larger cleanup slices with the loss-prevention guards active for remaining ECStore compatibility contracts now that broad compatibility passthroughs are fully closed. @@ -1753,14 +1763,24 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block | Expert | Status | Notes | |---|---|---| -| Quality/architecture | passed | S-015 removes obsolete KMS admin policy action variants after the handler fallback cleanup; API-042/API-043/API-044/API-045/API-046/API-047/API-048/API-049/API-050/API-051/API-052/API-053/API-054 narrow notify, S3 Select, OBS, IAM, Swift, heal, scanner, RustFS runtime, test, fuzz, lifecycle helper, harness, and RustFS runtime compatibility contracts without moving ECStore storage metadata ownership; G-011/G-012/G-013 add docs-only baselines for scheduler, placement/repair, and profiling/NUMA work; Issue #660 PR-08/PR-09 add read-only observability and topology contracts in storage-api only; PR-05/PR-07 add scheduler preservation tests and workload contracts; current API-055/SCH-001 adds a local storage concurrency provider only. | +| Quality/architecture | passed | S-015 removes obsolete KMS admin policy action variants after the handler fallback cleanup; API-042/API-043/API-044/API-045/API-046/API-047/API-048/API-049/API-050/API-051/API-052/API-053/API-054 narrow notify, S3 Select, OBS, IAM, Swift, heal, scanner, RustFS runtime, test, fuzz, lifecycle helper, harness, and RustFS runtime compatibility contracts without moving ECStore storage metadata ownership; G-011/G-012/G-013 add docs-only baselines for scheduler, placement/repair, and profiling/NUMA work; Issue #660 PR-08/PR-09 add read-only observability and topology contracts in storage-api only; PR-05/PR-07 add scheduler preservation tests and workload contracts; API-055/SCH-001 adds a local storage concurrency provider; current API-056/R-016 wires runtime observability and endpoint topology providers without moving implementation ownership. | | Migration preservation | passed | KMS endpoint URLs, query aliases, request bodies, response contracts, and dedicated `kms:*` authorization behavior are preserved; event builder call sites, ECStore event bridge conversion, restore event data, version IDs, metadata filtering, config read/save semantics, S3 Select store/error/buffer semantics, OBS metrics state reads, IAM config/notification/error semantics, Swift bucket metadata access, heal disk/resume/task behavior, scanner lifecycle/replication/data-usage behavior, RustFS startup/admin/app/storage runtime access, e2e/test/fuzz import behavior, lifecycle expiration/transition helper DTO field contracts, flattened harness and RustFS runtime scalar/secondary alias behavior, unchanged no-op handling, remove-event behavior, scheduler/readiness/placement/profiling runtime behavior, platform gates, missing/unknown capability states, placement/topology labels, scheduler thresholds, queue snapshot semantics, disk-read semaphore behavior, and admission behavior are preserved. | -| Testing/verification | passed | Focused compiles/tests, fuzz target compile, guards, formatting, diff hygiene, risk scan, and full `make pre-commit` passed for prior code slices; current Issue #660 API-055/SCH-001 slice uses storage concurrency tests, focused RustFS library check, migration guard, formatting, diff hygiene, and three-expert review. | +| Testing/verification | passed | Focused compiles/tests, fuzz target compile, guards, formatting, diff hygiene, risk scan, and full `make pre-commit` passed for prior code slices; current Issue #660 API-056/R-016 slice uses runtime capability provider tests, focused RustFS library check, migration and layer guards, formatting, diff hygiene, and three-expert review. | ## Verification Notes Passed before push: +- Issue #660 API-056/R-016 current slice: + - `cargo test -p rustfs --lib runtime_capabilities -- --nocapture`: passed. + - `cargo check -p rustfs --lib`: passed. + - `./scripts/check_architecture_migration_rules.sh`: passed. + - `./scripts/check_layer_dependencies.sh`: passed. + - `cargo fmt --all --check`: passed. + - `git diff --check`: passed. + - `make pre-commit`: passed. + - Three-expert review: passed. + - Issue #660 API-055/SCH-001 current slice: - `cargo test -p rustfs --lib storage::concurrency::manager::integration_tests -- --nocapture`: passed. - `cargo check -p rustfs --lib`: passed. diff --git a/docs/architecture/runtime-capability-contracts.md b/docs/architecture/runtime-capability-contracts.md index e2b5ac0e4..29eb6f56c 100644 --- a/docs/architecture/runtime-capability-contracts.md +++ b/docs/architecture/runtime-capability-contracts.md @@ -42,3 +42,20 @@ ECStore type leakage. exporter behavior changes are part of this contract slice. - Providers must map implementation failures into `CapabilitySnapshotError` before crossing the contract boundary. + +## RustFS Provider Slice + +`rustfs/src/runtime_capabilities.rs` wires the contracts to RustFS runtime +owners through read-only providers: + +- `RustFsObservabilitySnapshotProvider` maps current dial9, profiling, + memory-sampling, platform, allocator, eBPF, and NUMA capability state without + starting telemetry, profiling, allocator reclaim, or memory-observability + workers. +- `EndpointTopologySnapshotProvider` maps `EndpointServerPools` into pool, set, + and disk topology snapshots without changing endpoint construction, + placement, readiness, locks, or ECStore metadata. Local file endpoint paths are + intentionally not used as disk IDs or labels. + +Unsupported or unavailable runtime capabilities are reported as `unsupported` +or `unknown` contract states instead of activating fallback behavior. diff --git a/rustfs/src/lib.rs b/rustfs/src/lib.rs index 996aa98be..fddcd51ba 100644 --- a/rustfs/src/lib.rs +++ b/rustfs/src/lib.rs @@ -66,6 +66,7 @@ pub mod memory_observability; pub mod profiling; #[cfg(any(feature = "ftps", feature = "webdav", feature = "sftp"))] pub mod protocols; +pub mod runtime_capabilities; pub mod server; pub mod startup_entrypoint; pub mod startup_fs_guard; diff --git a/rustfs/src/runtime_capabilities.rs b/rustfs/src/runtime_capabilities.rs new file mode 100644 index 000000000..7b34bae55 --- /dev/null +++ b/rustfs/src/runtime_capabilities.rs @@ -0,0 +1,286 @@ +// 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 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::collections::BTreeMap; + +use rustfs_storage_api::{ + CapabilitySnapshotError, CapabilityStatus, DiskCapabilities, MemorySamplingState, ObservabilitySnapshot, + ObservabilitySnapshotProvider, PlatformSupport, TopologyCapabilities, TopologyDisk, TopologyLabels, TopologyPool, + TopologySet, TopologySnapshot, TopologySnapshotProvider, UserspaceProfilingCapability, +}; + +use crate::storage_compat::{Endpoint, EndpointServerPools}; + +const NOT_WIRED_INTO_RUNTIME: &str = "not wired into runtime"; +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"; + +#[derive(Debug, Default, Clone, Copy)] +pub struct RustFsObservabilitySnapshotProvider; + +#[async_trait::async_trait] +impl ObservabilitySnapshotProvider for RustFsObservabilitySnapshotProvider { + async fn observability_snapshot(&self) -> Result { + Ok(runtime_observability_snapshot()) + } +} + +pub fn runtime_observability_snapshot() -> ObservabilitySnapshot { + ObservabilitySnapshot { + runtime_telemetry: runtime_telemetry_status(), + userspace_profiling: UserspaceProfilingCapability { + cpu: cpu_profiling_status(), + memory: memory_profiling_status(), + continuous_cpu: cpu_profiling_status(), + periodic_cpu: cpu_profiling_status(), + }, + memory_sampling: MemorySamplingState { + process: CapabilityStatus::supported(), + system: CapabilityStatus::supported(), + cgroup: cgroup_memory_status(), + }, + platform: PlatformSupport { + target_triple: option_env!("TARGET").map(str::to_owned), + 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), + }, + } +} + +#[derive(Debug, Clone)] +pub struct EndpointTopologySnapshotProvider { + endpoint_pools: EndpointServerPools, +} + +impl EndpointTopologySnapshotProvider { + pub fn new(endpoint_pools: EndpointServerPools) -> Self { + Self { endpoint_pools } + } +} + +#[async_trait::async_trait] +impl TopologySnapshotProvider for EndpointTopologySnapshotProvider { + async fn topology_snapshot(&self) -> Result { + Ok(topology_snapshot_from_endpoint_pools(&self.endpoint_pools)) + } +} + +pub fn topology_snapshot_from_endpoint_pools(endpoint_pools: &EndpointServerPools) -> TopologySnapshot { + TopologySnapshot { + pools: endpoint_pools + .as_ref() + .iter() + .enumerate() + .map(|(pool_index, pool)| { + let sets = topology_sets_from_endpoints(pool_index, pool.drives_per_set, pool.endpoints.as_ref()); + TopologyPool { + pool_index, + pool_id: None, + labels: TopologyLabels::default(), + sets, + } + }) + .collect(), + capabilities: TopologyCapabilities { + profiling: cpu_profiling_status(), + numa: CapabilityStatus::unsupported().with_reason(NUMA_NOT_WIRED), + failure_domain_labels: CapabilityStatus::unknown().with_reason(FAILURE_DOMAIN_NOT_REPORTED), + media_labels: CapabilityStatus::unknown().with_reason(STORAGE_MEDIA_NOT_REPORTED), + }, + } +} + +fn topology_sets_from_endpoints(pool_index: usize, drives_per_set: usize, endpoints: &[Endpoint]) -> Vec { + let safe_drives_per_set = drives_per_set.max(1); + let mut sets = BTreeMap::>::new(); + + for (endpoint_index, endpoint) in endpoints.iter().enumerate() { + let set_index = non_negative_index(endpoint.set_idx).unwrap_or(endpoint_index / safe_drives_per_set); + let disk_index = non_negative_index(endpoint.disk_idx).unwrap_or(endpoint_index % safe_drives_per_set); + + sets.entry(set_index) + .or_default() + .push(topology_disk_from_endpoint(pool_index, set_index, disk_index, endpoint)); + } + + sets.into_iter() + .map(|(set_index, mut disks)| { + disks.sort_by_key(|disk| disk.disk_index); + TopologySet { + pool_index, + set_index, + set_id: None, + labels: TopologyLabels::default(), + disks, + } + }) + .collect() +} + +fn topology_disk_from_endpoint(pool_index: usize, set_index: usize, disk_index: usize, endpoint: &Endpoint) -> TopologyDisk { + TopologyDisk { + pool_index, + set_index, + disk_index, + disk_id: endpoint_disk_id(endpoint), + labels: endpoint_labels(endpoint), + capabilities: DiskCapabilities { + media_type: CapabilityStatus::unknown().with_reason(STORAGE_MEDIA_NOT_REPORTED), + failure_domain: CapabilityStatus::unknown().with_reason(FAILURE_DOMAIN_NOT_REPORTED), + numa: CapabilityStatus::unsupported().with_reason(NUMA_NOT_WIRED), + profiling: cpu_profiling_status(), + }, + } +} + +fn endpoint_disk_id(endpoint: &Endpoint) -> Option { + let host_port = endpoint.host_port(); + if host_port.is_empty() { None } else { Some(host_port) } +} + +fn endpoint_labels(endpoint: &Endpoint) -> TopologyLabels { + let mut additional = BTreeMap::new(); + additional.insert( + "endpoint_type".to_owned(), + if endpoint.url.scheme() == "file" { "path" } else { "url" }.to_owned(), + ); + additional.insert("local".to_owned(), endpoint.is_local.to_string()); + + TopologyLabels { + additional, + ..TopologyLabels::default() + } +} + +fn non_negative_index(index: i32) -> Option { + usize::try_from(index).ok() +} + +fn runtime_telemetry_status() -> CapabilityStatus { + if rustfs_obs::dial9::is_enabled() { + CapabilityStatus::supported() + } else { + CapabilityStatus::disabled() + } +} + +fn cpu_profiling_status() -> CapabilityStatus { + if cfg!(any(target_os = "linux", target_os = "macos")) { + CapabilityStatus::supported() + } else { + CapabilityStatus::unsupported().with_reason("userspace CPU profiling supports linux and macos targets") + } +} + +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") + } else { + CapabilityStatus::unsupported().with_reason("memory profiling supports linux gnu x86_64 targets") + } +} + +fn cgroup_memory_status() -> CapabilityStatus { + if cfg!(target_os = "linux") { + CapabilityStatus::supported() + } else { + CapabilityStatus::unsupported().with_reason("cgroup memory sampling supports linux targets") + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::storage_compat::{Endpoints, PoolEndpoints}; + use rustfs_storage_api::{CapabilityState, ObservabilitySnapshotProvider, TopologySnapshotProvider}; + + #[tokio::test] + async fn observability_provider_returns_platform_snapshot() { + let provider = RustFsObservabilitySnapshotProvider; + let snapshot = provider.observability_snapshot().await.expect("observability snapshot"); + + 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!(matches!( + snapshot.runtime_telemetry.state, + CapabilityState::Supported | CapabilityState::Disabled + )); + assert!( + snapshot + .platform + .allocator + .reason + .as_deref() + .unwrap_or_default() + .contains("backend=") + ); + } + + #[tokio::test] + async fn topology_provider_maps_endpoint_pools_to_sets_and_disks() { + let endpoint_pools = sample_endpoint_pools(); + let provider = EndpointTopologySnapshotProvider::new(endpoint_pools); + let snapshot = provider.topology_snapshot().await.expect("topology snapshot"); + + assert_eq!(snapshot.pools.len(), 1); + assert_eq!(snapshot.pools[0].sets.len(), 2); + assert_eq!(snapshot.pools[0].sets[0].disks.len(), 2); + assert_eq!(snapshot.pools[0].sets[1].disks.len(), 2); + assert_eq!(snapshot.pools[0].sets[0].disks[1].disk_index, 1); + assert_eq!( + snapshot.pools[0].sets[0].disks[0] + .labels + .additional + .get("endpoint_type") + .map(String::as_str), + Some("path") + ); + } + + #[test] + fn topology_snapshot_does_not_expose_local_paths() { + let endpoint_pools = sample_endpoint_pools(); + let snapshot = topology_snapshot_from_endpoint_pools(&endpoint_pools); + let encoded = serde_json::to_string(&snapshot).expect("serialize topology snapshot"); + + assert!(!encoded.contains("/tmp/rustfs-runtime-capability")); + } + + fn sample_endpoint_pools() -> EndpointServerPools { + let endpoints = (0..4) + .map(|index| { + let mut endpoint = + Endpoint::try_from(format!("/tmp/rustfs-runtime-capability-{index}").as_str()).expect("local endpoint"); + endpoint.set_pool_index(0); + endpoint.set_set_index(index / 2); + endpoint.set_disk_index(index % 2); + endpoint + }) + .collect::>(); + + EndpointServerPools::from(vec![PoolEndpoints { + legacy: false, + set_count: 2, + drives_per_set: 2, + endpoints: Endpoints::from(endpoints), + cmd_line: "/tmp/rustfs-runtime-capability-{0...3}".to_owned(), + platform: "OS: test | Arch: test".to_owned(), + }]) + } +} diff --git a/scripts/check_architecture_migration_rules.sh b/scripts/check_architecture_migration_rules.sh index bfcacb7ea..d0794c9be 100755 --- a/scripts/check_architecture_migration_rules.sh +++ b/scripts/check_architecture_migration_rules.sh @@ -280,6 +280,22 @@ require_source_line \ "crates/storage-api/src/lib.rs" \ "pub use topology::{" \ "storage-api public topology contract re-export" +require_source_line \ + "rustfs/src/lib.rs" \ + "pub mod runtime_capabilities;" \ + "RustFS runtime capability provider module" +require_source_contains \ + "rustfs/src/runtime_capabilities.rs" \ + "impl ObservabilitySnapshotProvider for RustFsObservabilitySnapshotProvider" \ + "RustFS observability snapshot provider implementation" +require_source_contains \ + "rustfs/src/runtime_capabilities.rs" \ + "impl TopologySnapshotProvider for EndpointTopologySnapshotProvider" \ + "RustFS endpoint topology snapshot provider implementation" +require_source_contains \ + "rustfs/src/runtime_capabilities.rs" \ + "pub fn topology_snapshot_from_endpoint_pools(endpoint_pools: &EndpointServerPools) -> TopologySnapshot" \ + "RustFS endpoint topology snapshot helper" ( cd "$ROOT_DIR"