mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-06 05:17:42 +00:00
feat: expose runtime workload owner snapshots (#3607)
This commit is contained in:
@@ -5,18 +5,19 @@ 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-scanner-replication-admission-snapshots`
|
||||
- Baseline: `origin/main` after `rustfs/rustfs#3605`
|
||||
(`00ca3b7c1c4ffbe98c0dd8530cb26b63e94c586a`).
|
||||
- Branch: `overtrue/arch-foreground-scanner-admission-snapshots`
|
||||
- Baseline: `origin/main` after `rustfs/rustfs#3606`
|
||||
(`7cb7aefc3bfd33cfaa46c718db15f00b0471fe88`).
|
||||
- PR type for this branch: `consumer-migration`
|
||||
- Runtime behavior changes: none.
|
||||
- Rust code changes: extend read-only workload admission snapshots from heal
|
||||
repair counters to replication runtime worker and queue counters.
|
||||
- Rust code changes: extend the RustFS workload admission registry from
|
||||
repair/replication snapshots to foreground-read, scanner, and metadata owner
|
||||
snapshots.
|
||||
- CI/script changes: extend migration guard coverage for RustFS workload
|
||||
admission provider implementations.
|
||||
- Docs changes: add RustFS replication provider notes to
|
||||
- Docs changes: add RustFS runtime owner provider notes to
|
||||
[`workload-admission-contracts.md`](workload-admission-contracts.md) and
|
||||
record the API-058/R-018 provider slice.
|
||||
record the API-059/R-019 provider slice.
|
||||
|
||||
## Phase 0 Tasks
|
||||
|
||||
@@ -198,6 +199,21 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block
|
||||
check, migration and layer guards, formatting, diff hygiene, risk scan, and
|
||||
three-expert review.
|
||||
|
||||
- [x] `API-059/R-019` Expose RustFS runtime owner admission snapshots.
|
||||
- Completed slice: extend the RustFS workload admission provider to map
|
||||
foreground-read disk permit state, scanner active work units, and bucket
|
||||
metadata runtime initialization into the workload registry.
|
||||
- Acceptance: RustFS-level workload admission snapshots expose existing
|
||||
foreground-read, scanner, and metadata owner state without changing
|
||||
admission, queueing, scanner scheduling, metadata loading, metadata locks,
|
||||
or object write behavior.
|
||||
- Must preserve: disk-read semaphore acquisition, scanner cycle scheduling,
|
||||
bucket metadata initialization and loading, object write paths, request
|
||||
guards, and queue behavior.
|
||||
- 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.
|
||||
- Acceptance: `./scripts/check_architecture_migration_rules.sh` parses the
|
||||
allowed PR types from [`crate-boundaries.md`](crate-boundaries.md) and fails
|
||||
|
||||
@@ -92,3 +92,23 @@ snapshot from the existing replication pool and queue statistics:
|
||||
This is an observation surface only. Replication admission, queue channel
|
||||
capacity, worker resize behavior, MRF handling, target dispatch, and resync
|
||||
behavior are unchanged.
|
||||
|
||||
## RustFS Runtime Owner Snapshot Extraction
|
||||
|
||||
The RustFS integration layer now extends the workload admission registry with
|
||||
additional read-only owner mappings:
|
||||
|
||||
- `ForegroundRead` reuses the storage `ConcurrencyManager` disk-read permit
|
||||
snapshot so the RustFS-level provider exposes the same active and limit
|
||||
counts as the storage-local provider.
|
||||
- `Scanner` reports the existing scanner active work-unit counter. When the
|
||||
counter is zero, the snapshot remains `Unknown` because the current counter
|
||||
cannot distinguish an idle scanner from a scanner that has not initialized.
|
||||
- `Metadata` reports `Open` once the bucket metadata runtime handle is
|
||||
available, and `Unknown` before initialization.
|
||||
- `ForegroundWrite` remains `Unknown` until a write-specific admission owner
|
||||
exposes a read-only surface.
|
||||
|
||||
This is an observation surface only. Disk-read permit acquisition, scanner
|
||||
cycle scheduling, bucket metadata loading, metadata locks, object write paths,
|
||||
and queue behavior are unchanged.
|
||||
|
||||
@@ -12,7 +12,7 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
pub(crate) use rustfs_ecstore::bucket::metadata_sys::init_bucket_metadata_sys;
|
||||
pub(crate) use rustfs_ecstore::bucket::metadata_sys::{get_global_bucket_metadata_sys, init_bucket_metadata_sys};
|
||||
pub(crate) use rustfs_ecstore::bucket::migration::{try_migrate_bucket_metadata, try_migrate_iam_config};
|
||||
pub(crate) use rustfs_ecstore::bucket::replication::{
|
||||
GLOBAL_REPLICATION_STATS, get_global_replication_pool, init_background_replication,
|
||||
|
||||
@@ -17,12 +17,15 @@ use rustfs_concurrency::{
|
||||
WorkloadClass,
|
||||
};
|
||||
|
||||
use crate::storage_compat::{GLOBAL_REPLICATION_STATS, get_global_replication_pool};
|
||||
use crate::storage::concurrency::get_concurrency_manager;
|
||||
use crate::storage_compat::{GLOBAL_REPLICATION_STATS, get_global_bucket_metadata_sys, get_global_replication_pool};
|
||||
|
||||
const BUCKET_METADATA_RUNTIME_NOT_INITIALIZED: &str = "bucket metadata runtime not initialized";
|
||||
const HEAL_MANAGER_NOT_INITIALIZED: &str = "heal manager not initialized";
|
||||
const NOT_EXPOSED_BY_PROVIDER: &str = "not exposed by RustFS workload admission provider";
|
||||
const REPLICATION_RUNTIME_NOT_INITIALIZED: &str = "replication runtime not initialized";
|
||||
const REPLICATION_QUEUE_STATS_UNAVAILABLE: &str = "replication queue stats unavailable";
|
||||
const SCANNER_ACTIVITY_IDLE_OR_NOT_INITIALIZED: &str = "scanner activity idle or not initialized";
|
||||
|
||||
#[derive(Debug, Default, Clone, Copy)]
|
||||
pub struct RustFsWorkloadAdmissionSnapshotProvider;
|
||||
@@ -38,6 +41,9 @@ pub fn workload_admission_registry_snapshot() -> WorkloadAdmissionRegistrySnapsh
|
||||
.iter()
|
||||
.copied()
|
||||
.map(|class| match class {
|
||||
WorkloadClass::ForegroundRead => foreground_read_workload_admission_snapshot(),
|
||||
WorkloadClass::Metadata => metadata_workload_admission_snapshot(),
|
||||
WorkloadClass::Scanner => scanner_workload_admission_snapshot(),
|
||||
WorkloadClass::Repair => repair_workload_admission_snapshot(),
|
||||
WorkloadClass::Replication => replication_workload_admission_snapshot(),
|
||||
class => WorkloadAdmissionSnapshot::new(class, AdmissionState::Unknown).with_reason(NOT_EXPOSED_BY_PROVIDER),
|
||||
@@ -47,6 +53,54 @@ pub fn workload_admission_registry_snapshot() -> WorkloadAdmissionRegistrySnapsh
|
||||
WorkloadAdmissionRegistrySnapshot::new(entries)
|
||||
}
|
||||
|
||||
pub fn foreground_read_workload_admission_snapshot() -> WorkloadAdmissionSnapshot {
|
||||
get_concurrency_manager().get_object_admission_snapshot()
|
||||
}
|
||||
|
||||
pub fn metadata_workload_admission_snapshot() -> WorkloadAdmissionSnapshot {
|
||||
metadata_workload_admission_snapshot_from_initialized(get_global_bucket_metadata_sys().is_some())
|
||||
}
|
||||
|
||||
fn metadata_workload_admission_snapshot_from_initialized(runtime_initialized: bool) -> WorkloadAdmissionSnapshot {
|
||||
let state = if runtime_initialized {
|
||||
AdmissionState::Open
|
||||
} else {
|
||||
AdmissionState::Unknown
|
||||
};
|
||||
|
||||
let snapshot = WorkloadAdmissionSnapshot::new(WorkloadClass::Metadata, state);
|
||||
|
||||
if runtime_initialized {
|
||||
snapshot
|
||||
} else {
|
||||
snapshot.with_reason(BUCKET_METADATA_RUNTIME_NOT_INITIALIZED)
|
||||
}
|
||||
}
|
||||
|
||||
pub fn scanner_workload_admission_snapshot() -> WorkloadAdmissionSnapshot {
|
||||
scanner_workload_admission_snapshot_from_activity(rustfs_scanner::current_scanner_activity())
|
||||
}
|
||||
|
||||
fn scanner_workload_admission_snapshot_from_activity(active: u64) -> WorkloadAdmissionSnapshot {
|
||||
let state = if active > 0 {
|
||||
AdmissionState::Open
|
||||
} else {
|
||||
AdmissionState::Unknown
|
||||
};
|
||||
|
||||
let snapshot = WorkloadAdmissionSnapshot::new(WorkloadClass::Scanner, state).with_counts(
|
||||
Some(u64_to_usize_saturated(active)),
|
||||
None,
|
||||
None,
|
||||
);
|
||||
|
||||
if state == AdmissionState::Unknown {
|
||||
snapshot.with_reason(SCANNER_ACTIVITY_IDLE_OR_NOT_INITIALIZED)
|
||||
} else {
|
||||
snapshot
|
||||
}
|
||||
}
|
||||
|
||||
pub fn repair_workload_admission_snapshot() -> WorkloadAdmissionSnapshot {
|
||||
repair_workload_admission_snapshot_from_counts(
|
||||
rustfs_heal::get_heal_manager().is_some(),
|
||||
@@ -137,6 +191,59 @@ fn i32_to_usize_saturated(value: i32) -> usize {
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn foreground_read_snapshot_uses_storage_concurrency_admission() {
|
||||
let snapshot = foreground_read_workload_admission_snapshot();
|
||||
|
||||
assert_eq!(snapshot.class, WorkloadClass::ForegroundRead);
|
||||
assert_ne!(snapshot.state, AdmissionState::Unknown);
|
||||
assert!(snapshot.active.is_some());
|
||||
assert!(snapshot.limit.is_some());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn metadata_snapshot_reports_initialized_runtime() {
|
||||
let snapshot = metadata_workload_admission_snapshot_from_initialized(true);
|
||||
|
||||
assert_eq!(snapshot.class, WorkloadClass::Metadata);
|
||||
assert_eq!(snapshot.state, AdmissionState::Open);
|
||||
assert_eq!(snapshot.active, None);
|
||||
assert_eq!(snapshot.queued, None);
|
||||
assert_eq!(snapshot.limit, None);
|
||||
assert_eq!(snapshot.reason, None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn metadata_snapshot_is_unknown_before_runtime_initializes() {
|
||||
let snapshot = metadata_workload_admission_snapshot_from_initialized(false);
|
||||
|
||||
assert_eq!(snapshot.class, WorkloadClass::Metadata);
|
||||
assert_eq!(snapshot.state, AdmissionState::Unknown);
|
||||
assert_eq!(snapshot.reason.as_deref(), Some(BUCKET_METADATA_RUNTIME_NOT_INITIALIZED));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn scanner_snapshot_reports_active_work_units() {
|
||||
let snapshot = scanner_workload_admission_snapshot_from_activity(5);
|
||||
|
||||
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.reason, None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn scanner_snapshot_is_unknown_when_idle_or_uninitialized() {
|
||||
let snapshot = scanner_workload_admission_snapshot_from_activity(0);
|
||||
|
||||
assert_eq!(snapshot.class, WorkloadClass::Scanner);
|
||||
assert_eq!(snapshot.state, AdmissionState::Unknown);
|
||||
assert_eq!(snapshot.active, Some(0));
|
||||
assert_eq!(snapshot.reason.as_deref(), Some(SCANNER_ACTIVITY_IDLE_OR_NOT_INITIALIZED));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn repair_snapshot_reports_heal_counters() {
|
||||
let snapshot = repair_workload_admission_snapshot_from_counts(true, 2, 3);
|
||||
@@ -206,9 +313,10 @@ mod tests {
|
||||
|
||||
assert!(registry.get(WorkloadClass::Repair).is_some());
|
||||
assert!(registry.get(WorkloadClass::Replication).is_some());
|
||||
assert_eq!(
|
||||
assert_ne!(
|
||||
registry.get(WorkloadClass::ForegroundRead).map(|snapshot| snapshot.state),
|
||||
Some(AdmissionState::Unknown)
|
||||
);
|
||||
assert_eq!(registry.get(WorkloadClass::Scanner).and_then(|snapshot| snapshot.active), Some(0));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -304,6 +304,30 @@ 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 foreground_read_workload_admission_snapshot() -> WorkloadAdmissionSnapshot" \
|
||||
"RustFS foreground-read workload admission snapshot helper"
|
||||
require_source_contains \
|
||||
"rustfs/src/workload_admission.rs" \
|
||||
"WorkloadClass::ForegroundRead => foreground_read_workload_admission_snapshot()" \
|
||||
"RustFS foreground-read workload admission class mapping"
|
||||
require_source_contains \
|
||||
"rustfs/src/workload_admission.rs" \
|
||||
"pub fn metadata_workload_admission_snapshot() -> WorkloadAdmissionSnapshot" \
|
||||
"RustFS metadata workload admission snapshot helper"
|
||||
require_source_contains \
|
||||
"rustfs/src/workload_admission.rs" \
|
||||
"WorkloadClass::Metadata => metadata_workload_admission_snapshot()" \
|
||||
"RustFS metadata workload admission class mapping"
|
||||
require_source_contains \
|
||||
"rustfs/src/workload_admission.rs" \
|
||||
"pub fn scanner_workload_admission_snapshot() -> WorkloadAdmissionSnapshot" \
|
||||
"RustFS scanner workload admission snapshot helper"
|
||||
require_source_contains \
|
||||
"rustfs/src/workload_admission.rs" \
|
||||
"WorkloadClass::Scanner => scanner_workload_admission_snapshot()" \
|
||||
"RustFS scanner workload admission class mapping"
|
||||
require_source_contains \
|
||||
"rustfs/src/workload_admission.rs" \
|
||||
"pub fn repair_workload_admission_snapshot() -> WorkloadAdmissionSnapshot" \
|
||||
|
||||
Reference in New Issue
Block a user