feat: wire runtime capability snapshot providers (#3604)

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.

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.

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;

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<ObservabilitySnapshot, CapabilitySnapshotError> {
+        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<TopologySnapshot, CapabilitySnapshotError> {
+        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<TopologySet> {
+    let safe_drives_per_set = drives_per_set.max(1);
+    let mut sets = BTreeMap::<usize, Vec<TopologyDisk>>::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<String> {
+    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> {
+    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::<Vec<_>>();
+
+        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(),
+        }])
+    }
+}

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"