feat: expose local scheduler admission snapshots (#3603)

rustfs/src/storage/concurrency/manager.rs

@@ -19,7 +19,10 @@ use super::io_schedule::{
     get_advanced_buffer_size,
 };
 use super::request_guard::GetObjectGuard;
-use rustfs_concurrency::GetObjectQueueSnapshot;
+use rustfs_concurrency::{
+    AdmissionState, GetObjectQueueSnapshot, WorkloadAdmissionRegistrySnapshot, WorkloadAdmissionSnapshot,
+    WorkloadAdmissionSnapshotProvider, WorkloadClass,
+};
 use rustfs_config::{KI_B, MI_B};
 use rustfs_io_core::BytesPool;
 use rustfs_io_core::io_profile::{AccessPattern, IoPatternDetector, StorageMedia, detect_storage_media};
@@ -521,10 +524,7 @@ impl ConcurrencyManager {
     ///
     /// Returns information about permit usage and waiting requests.
     pub fn io_queue_status(&self) -> IoQueueStatus {
-        let snapshot = GetObjectQueueSnapshot::from_available_permits(
-            self.scheduler_config.max_concurrent_reads,
-            self.disk_read_semaphore.available_permits(),
-        );
+        let snapshot = self.get_object_queue_snapshot();
 
         IoQueueStatus {
             total_permits: snapshot.total_permits,
@@ -539,6 +539,53 @@ impl ConcurrencyManager {
         }
     }
 
+    /// Get a read-only snapshot of local GetObject disk-read admission.
+    pub fn get_object_queue_snapshot(&self) -> GetObjectQueueSnapshot {
+        GetObjectQueueSnapshot::from_available_permits(
+            self.scheduler_config.max_concurrent_reads,
+            self.disk_read_semaphore.available_permits(),
+        )
+    }
+
+    /// Get a read-only workload admission snapshot for foreground reads.
+    pub fn get_object_admission_snapshot(&self) -> WorkloadAdmissionSnapshot {
+        let snapshot = self.get_object_queue_snapshot();
+        let state = if snapshot.total_permits == 0 {
+            AdmissionState::Disabled
+        } else if snapshot.permits_available() == 0 {
+            AdmissionState::Saturated
+        } else {
+            AdmissionState::Open
+        };
+
+        let admission = WorkloadAdmissionSnapshot::new(WorkloadClass::ForegroundRead, state).with_counts(
+            Some(snapshot.permits_in_use),
+            None,
+            Some(snapshot.total_permits),
+        );
+
+        match state {
+            AdmissionState::Disabled => admission.with_reason("disk read permits disabled"),
+            AdmissionState::Saturated => admission.with_reason("all disk read permits are in use"),
+            _ => admission,
+        }
+    }
+
+    /// Get a read-only workload admission registry snapshot for local storage concurrency.
+    pub fn workload_admission_registry_snapshot(&self) -> WorkloadAdmissionRegistrySnapshot {
+        let entries = WorkloadClass::REQUIRED
+            .iter()
+            .copied()
+            .map(|class| match class {
+                WorkloadClass::ForegroundRead => self.get_object_admission_snapshot(),
+                class => WorkloadAdmissionSnapshot::new(class, AdmissionState::Unknown)
+                    .with_reason("not exposed by storage concurrency manager"),
+            })
+            .collect();
+
+        WorkloadAdmissionRegistrySnapshot::new(entries)
+    }
+
     /// Acquire a disk read permit with priority awareness.
     ///
     /// When priority scheduling is enabled, this method logs the priority
@@ -573,6 +620,12 @@ impl ConcurrencyManager {
     }
 }
 
+impl WorkloadAdmissionSnapshotProvider for ConcurrencyManager {
+    fn workload_admission_snapshot(&self) -> WorkloadAdmissionRegistrySnapshot {
+        self.workload_admission_registry_snapshot()
+    }
+}
+
 impl Default for ConcurrencyManager {
     fn default() -> Self {
         Self::new()
@@ -617,6 +670,54 @@ mod integration_tests {
         assert_eq!(status.low_priority_waiting, 0);
     }
 
+    #[tokio::test]
+    #[serial]
+    async fn test_concurrency_manager_workload_admission_snapshot_tracks_disk_read_permits() {
+        let manager = ConcurrencyManager::new();
+        let initial = manager.get_object_admission_snapshot();
+
+        assert_eq!(initial.class, WorkloadClass::ForegroundRead);
+        assert_eq!(initial.state, AdmissionState::Open);
+        assert_eq!(initial.active, Some(0));
+        assert_eq!(initial.queued, None);
+        assert_eq!(initial.limit, Some(manager.scheduler_config().max_concurrent_reads));
+
+        let permit = manager.acquire_disk_read_permit().await;
+        assert!(permit.is_ok());
+        let _permit = match permit {
+            Ok(permit) => permit,
+            Err(error) => panic!("disk read permit acquisition failed: {error}"),
+        };
+        let snapshot = manager.get_object_admission_snapshot();
+
+        assert_eq!(snapshot.active, Some(1));
+        assert_eq!(snapshot.limit, Some(manager.scheduler_config().max_concurrent_reads));
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn test_concurrency_manager_workload_admission_registry_covers_required_classes() {
+        let manager = ConcurrencyManager::new();
+        let registry = manager.workload_admission_registry_snapshot();
+        let provider: &dyn WorkloadAdmissionSnapshotProvider = &manager;
+        let trait_registry = provider.workload_admission_snapshot();
+
+        assert_eq!(registry.entries(), trait_registry.entries());
+        assert_eq!(registry.entries().len(), WorkloadClass::REQUIRED.len());
+        for class in WorkloadClass::REQUIRED {
+            assert!(registry.get(class).is_some(), "missing workload class {class}");
+        }
+
+        assert_eq!(
+            registry.get(WorkloadClass::ForegroundRead).map(|snapshot| snapshot.state),
+            Some(AdmissionState::Open)
+        );
+        assert_eq!(
+            registry.get(WorkloadClass::Scanner).map(|snapshot| snapshot.state),
+            Some(AdmissionState::Unknown)
+        );
+    }
+
     #[tokio::test]
     #[serial]
     async fn test_concurrency_manager_disk_read_permit() {