feat: expose heal workload admission snapshot (#3605)

This commit is contained in:
安正超
2026-06-19 10:39:09 +08:00
committed by GitHub
parent 8cf3c0bfbd
commit 00ca3b7c1c
5 changed files with 180 additions and 10 deletions
+23 -10
View File
@@ -5,18 +5,18 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block
## Current Context ## Current Context
- Issue: [`rustfs/backlog#660`](https://github.com/rustfs/backlog/issues/660) - Issue: [`rustfs/backlog#660`](https://github.com/rustfs/backlog/issues/660)
- Branch: `overtrue/arch-runtime-capability-snapshots` - Branch: `overtrue/arch-workload-admission-owner-snapshots`
- Baseline: `origin/main` after `rustfs/rustfs#3603` - Baseline: `origin/main` after `rustfs/rustfs#3590`
(`b106b628c164bf6cdca9e71ff5118711f99d4ab0`). (`8cf3c0bf0100d9d2234f08b48df31fd73b92ed43`).
- PR type for this branch: `consumer-migration` - PR type for this branch: `consumer-migration`
- Runtime behavior changes: none. - Runtime behavior changes: none.
- Rust code changes: wire read-only observability and endpoint-topology - Rust code changes: expose read-only repair workload admission snapshots from
snapshot providers in the RustFS runtime. existing heal runtime counters.
- CI/script changes: extend migration guard coverage for RustFS runtime - CI/script changes: extend migration guard coverage for RustFS workload
capability provider implementations. admission provider implementations.
- Docs changes: add RustFS provider notes to - Docs changes: add RustFS heal provider notes to
[`runtime-capability-contracts.md`](runtime-capability-contracts.md) and [`workload-admission-contracts.md`](workload-admission-contracts.md) and
record the API-056/R-016 provider slice. record the API-057/R-017 provider slice.
## Phase 0 Tasks ## Phase 0 Tasks
@@ -170,6 +170,19 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block
check, migration and layer guards, formatting, diff hygiene, risk scan, and check, migration and layer guards, formatting, diff hygiene, risk scan, and
three-expert review. three-expert review.
- [x] `API-057/R-017` Expose heal repair admission snapshot.
- Completed slice: implement a RustFS workload admission snapshot provider
that maps existing heal active-task and queue-length counters to the
`Repair` workload class.
- Acceptance: repair admission state is observable through the
`rustfs-concurrency` workload snapshot contract without changing heal
queueing, scheduling, retry, priority merge/drop, or repair behavior.
- Must preserve: heal request admission, queue capacity, scheduler wakeups,
task retry handling, active-task accounting, and repair execution.
- Verification: focused workload admission 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. - [x] `TEST-PRTYPE-001` Check PR type enum consistency.
- Acceptance: `./scripts/check_architecture_migration_rules.sh` parses the - Acceptance: `./scripts/check_architecture_migration_rules.sh` parses the
allowed PR types from [`crate-boundaries.md`](crate-boundaries.md) and fails allowed PR types from [`crate-boundaries.md`](crate-boundaries.md) and fails
@@ -59,3 +59,19 @@ The RustFS storage `ConcurrencyManager` now implements
This is an observation surface only. Permit acquisition, priority assignment, This is an observation surface only. Permit acquisition, priority assignment,
buffer sizing, storage media detection, request guards, and queue behavior are buffer sizing, storage media detection, request guards, and queue behavior are
unchanged. unchanged.
## Heal Repair Snapshot Extraction
The RustFS integration layer now exposes a read-only repair admission snapshot
from the heal runtime counters:
- `Repair` reports the current heal active task count.
- `queued` reports the current heal queue length.
- `limit` remains `None` because the configured heal queue and concurrency
limits live behind the async heal manager state.
- Other workload classes remain `Unknown` in this provider until their owning
runtime components expose read-only status.
This is an observation surface only. Heal request admission, queue capacity,
priority merge/drop policy, task scheduling, retry handling, and repair
behavior are unchanged.
+1
View File
@@ -83,6 +83,7 @@ pub(crate) mod table_catalog;
pub mod tls; pub mod tls;
pub mod update; pub mod update;
pub mod version; pub mod version;
pub mod workload_admission;
// Re-export from rustfs_utils so that config sub-modules can use // Re-export from rustfs_utils so that config sub-modules can use
// `crate::apply_external_env_compat` without breaking. // `crate::apply_external_env_compat` without breaking.
+124
View File
@@ -0,0 +1,124 @@
// 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 rustfs_concurrency::{
AdmissionState, WorkloadAdmissionRegistrySnapshot, WorkloadAdmissionSnapshot, WorkloadAdmissionSnapshotProvider,
WorkloadClass,
};
const HEAL_MANAGER_NOT_INITIALIZED: &str = "heal manager not initialized";
const NOT_EXPOSED_BY_PROVIDER: &str = "not exposed by RustFS workload admission provider";
#[derive(Debug, Default, Clone, Copy)]
pub struct RustFsWorkloadAdmissionSnapshotProvider;
impl WorkloadAdmissionSnapshotProvider for RustFsWorkloadAdmissionSnapshotProvider {
fn workload_admission_snapshot(&self) -> WorkloadAdmissionRegistrySnapshot {
workload_admission_registry_snapshot()
}
}
pub fn workload_admission_registry_snapshot() -> WorkloadAdmissionRegistrySnapshot {
let entries = WorkloadClass::REQUIRED
.iter()
.copied()
.map(|class| match class {
WorkloadClass::Repair => repair_workload_admission_snapshot(),
class => WorkloadAdmissionSnapshot::new(class, AdmissionState::Unknown).with_reason(NOT_EXPOSED_BY_PROVIDER),
})
.collect();
WorkloadAdmissionRegistrySnapshot::new(entries)
}
pub fn repair_workload_admission_snapshot() -> WorkloadAdmissionSnapshot {
repair_workload_admission_snapshot_from_counts(
rustfs_heal::get_heal_manager().is_some(),
rustfs_heal::current_heal_active_tasks(),
rustfs_heal::current_heal_queue_length(),
)
}
fn repair_workload_admission_snapshot_from_counts(
manager_initialized: bool,
active: u64,
queued: u64,
) -> WorkloadAdmissionSnapshot {
let state = if manager_initialized || active > 0 || queued > 0 {
AdmissionState::Open
} else {
AdmissionState::Unknown
};
let snapshot = WorkloadAdmissionSnapshot::new(WorkloadClass::Repair, state).with_counts(
Some(u64_to_usize_saturated(active)),
Some(u64_to_usize_saturated(queued)),
None,
);
if state == AdmissionState::Unknown {
snapshot.with_reason(HEAL_MANAGER_NOT_INITIALIZED)
} else {
snapshot
}
}
fn u64_to_usize_saturated(value: u64) -> usize {
usize::try_from(value).unwrap_or(usize::MAX)
}
#[cfg(test)]
mod tests {
use super::*;
#[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.active, Some(2));
assert_eq!(snapshot.queued, Some(3));
assert_eq!(snapshot.limit, None);
assert_eq!(snapshot.reason, None);
}
#[test]
fn repair_snapshot_is_unknown_before_heal_manager_initializes() {
let snapshot = repair_workload_admission_snapshot_from_counts(false, 0, 0);
assert_eq!(snapshot.class, WorkloadClass::Repair);
assert_eq!(snapshot.state, AdmissionState::Unknown);
assert_eq!(snapshot.active, Some(0));
assert_eq!(snapshot.queued, Some(0));
assert_eq!(snapshot.reason.as_deref(), Some(HEAL_MANAGER_NOT_INITIALIZED));
}
#[test]
fn provider_covers_required_classes() {
let provider = RustFsWorkloadAdmissionSnapshotProvider;
let registry = provider.workload_admission_snapshot();
assert_eq!(registry.entries().len(), WorkloadClass::REQUIRED.len());
for class in WorkloadClass::REQUIRED {
assert!(registry.get(class).is_some(), "missing workload class {class}");
}
assert!(registry.get(WorkloadClass::Repair).is_some());
assert_eq!(
registry.get(WorkloadClass::ForegroundRead).map(|snapshot| snapshot.state),
Some(AdmissionState::Unknown)
);
}
}
@@ -284,6 +284,10 @@ require_source_line \
"rustfs/src/lib.rs" \ "rustfs/src/lib.rs" \
"pub mod runtime_capabilities;" \ "pub mod runtime_capabilities;" \
"RustFS runtime capability provider module" "RustFS runtime capability provider module"
require_source_line \
"rustfs/src/lib.rs" \
"pub mod workload_admission;" \
"RustFS workload admission provider module"
require_source_contains \ require_source_contains \
"rustfs/src/runtime_capabilities.rs" \ "rustfs/src/runtime_capabilities.rs" \
"impl ObservabilitySnapshotProvider for RustFsObservabilitySnapshotProvider" \ "impl ObservabilitySnapshotProvider for RustFsObservabilitySnapshotProvider" \
@@ -296,6 +300,18 @@ require_source_contains \
"rustfs/src/runtime_capabilities.rs" \ "rustfs/src/runtime_capabilities.rs" \
"pub fn topology_snapshot_from_endpoint_pools(endpoint_pools: &EndpointServerPools) -> TopologySnapshot" \ "pub fn topology_snapshot_from_endpoint_pools(endpoint_pools: &EndpointServerPools) -> TopologySnapshot" \
"RustFS endpoint topology snapshot helper" "RustFS endpoint topology snapshot helper"
require_source_contains \
"rustfs/src/workload_admission.rs" \
"impl WorkloadAdmissionSnapshotProvider for RustFsWorkloadAdmissionSnapshotProvider" \
"RustFS workload admission snapshot provider implementation"
require_source_contains \
"rustfs/src/workload_admission.rs" \
"pub fn repair_workload_admission_snapshot() -> WorkloadAdmissionSnapshot" \
"RustFS repair workload admission snapshot helper"
require_source_contains \
"rustfs/src/workload_admission.rs" \
"WorkloadClass::Repair => repair_workload_admission_snapshot()" \
"RustFS repair workload admission class mapping"
( (
cd "$ROOT_DIR" cd "$ROOT_DIR"