[PATCH perl-rs v2 16/40] pve-rs: resource-scheduling: expose auto rebalancing methods

[Daniel Kral <d.kral@proxmox.com>]

These methods expose the auto rebalancing methods of both the static and
dynamic scheduler.

As Scheduler::score_best_balancing_migration_candidates{,_topsis}()
takes a possible very large list of migration candidates, the binding
takes a more compact representation, which reduces the size that needs
to be generated on the caller's side and therefore the runtime of the
serialization from Perl to Rust.

Additionally, while decomposing the compact representation the input
data is validated since the underlying scoring methods do not further
validate whether their input is consistent with the cluster usage.

The method names score_best_balancing_migration_candidates{,_topsis}()
are chosen deliberately, so that future extensions can implement
score_best_balancing_migrations{,_topsis}(), which might allow to score
migrations without providing the candidates.

Signed-off-by: Daniel Kral <d.kral@proxmox.com>
---
changes v1 -> v2:
- improve patch message and documentation
- move to the end of the perl-rs changes, which makes it more consistent
  with the change order in pve-ha-manager as well
- uses `UsageAggregator` now to discern how usages are accumulated
- s/generate_migration_candidates_from
   /decompose_compact_migration_candidates
- make the decomposition of compact migration candidates more robust and
  do not use any unwraps or other causes of panic but the Mutex guard
  unwrap

 .../resource_scheduling/pve_dynamic.rs        | 57 +++++++++++-
 .../resource_scheduling/pve_static.rs         | 56 +++++++++++-
 .../bindings/resource_scheduling/resource.rs  | 88 ++++++++++++++++++-
 .../src/bindings/resource_scheduling/usage.rs | 15 ++++
 4 files changed, 211 insertions(+), 5 deletions(-)

diff --git a/pve-rs/src/bindings/resource_scheduling/pve_dynamic.rs b/pve-rs/src/bindings/resource_scheduling/pve_dynamic.rs
index 5b4373e..26f36d1 100644
--- a/pve-rs/src/bindings/resource_scheduling/pve_dynamic.rs
+++ b/pve-rs/src/bindings/resource_scheduling/pve_dynamic.rs
@@ -14,10 +14,15 @@ pub mod pve_rs_resource_scheduling_dynamic {
     use perlmod::Value;
     use proxmox_resource_scheduling::node::NodeStats;
     use proxmox_resource_scheduling::resource::ResourceStats;
+    use proxmox_resource_scheduling::scheduler::ScoredMigration;
     use proxmox_resource_scheduling::usage::Usage;
 
-    use crate::bindings::resource_scheduling::resource::PveResource;
-    use crate::bindings::resource_scheduling::usage::StartingAsStartedResourceAggregator;
+    use crate::bindings::resource_scheduling::resource::{
+        CompactMigrationCandidate, PveResource, decompose_compact_migration_candidates,
+    };
+    use crate::bindings::resource_scheduling::usage::{
+        IdentityAggregator, StartingAsStartedResourceAggregator,
+    };
 
     perlmod::declare_magic!(Box<Scheduler> : &Scheduler as "PVE::RS::ResourceScheduling::Dynamic");
 
@@ -157,6 +162,54 @@ pub mod pve_rs_resource_scheduling_dynamic {
         usage.remove_resource(sid);
     }
 
+    /// Method: Returns the load imbalance among the nodes.
+    ///
+    /// See [`proxmox_resource_scheduling::scheduler::Scheduler::node_imbalance`].
+    #[export]
+    pub fn calculate_node_imbalance(#[try_from_ref] this: &Scheduler) -> f64 {
+        let usage = this.inner.lock().unwrap();
+
+        usage.to_scheduler::<IdentityAggregator>().node_imbalance()
+    }
+
+    /// Method: Scores the given migration `candidates` by the best node imbalance improvement with
+    /// exhaustive search.
+    ///
+    /// See [`proxmox_resource_scheduling::scheduler::Scheduler::score_best_balancing_migration_candidates`].
+    #[export]
+    pub fn score_best_balancing_migration_candidates(
+        #[try_from_ref] this: &Scheduler,
+        candidates: Vec<CompactMigrationCandidate>,
+        limit: usize,
+    ) -> Result<Vec<ScoredMigration>, Error> {
+        let usage = this.inner.lock().unwrap();
+
+        let candidates = decompose_compact_migration_candidates(&usage, candidates)?;
+
+        Ok(usage
+            .to_scheduler::<IdentityAggregator>()
+            .score_best_balancing_migration_candidates(candidates, limit))
+    }
+
+    /// Method: Scores the given migration `candidates` by the best node imbalance improvement with
+    /// the TOPSIS method.
+    ///
+    /// See [`proxmox_resource_scheduling::scheduler::Scheduler::score_best_balancing_migration_candidates_topsis`].
+    #[export]
+    pub fn score_best_balancing_migration_candidates_topsis(
+        #[try_from_ref] this: &Scheduler,
+        candidates: Vec<CompactMigrationCandidate>,
+        limit: usize,
+    ) -> Result<Vec<ScoredMigration>, Error> {
+        let usage = this.inner.lock().unwrap();
+
+        let candidates = decompose_compact_migration_candidates(&usage, candidates)?;
+
+        usage
+            .to_scheduler::<IdentityAggregator>()
+            .score_best_balancing_migration_candidates_topsis(&candidates, limit)
+    }
+
     /// Method: Scores nodes to start a resource with the usage statistics `resource_stats` on.
     ///
     /// See [`proxmox_resource_scheduling::scheduler::Scheduler::score_nodes_to_start_resource`].
diff --git a/pve-rs/src/bindings/resource_scheduling/pve_static.rs b/pve-rs/src/bindings/resource_scheduling/pve_static.rs
index e2756db..7924889 100644
--- a/pve-rs/src/bindings/resource_scheduling/pve_static.rs
+++ b/pve-rs/src/bindings/resource_scheduling/pve_static.rs
@@ -14,10 +14,14 @@ pub mod pve_rs_resource_scheduling_static {
     use perlmod::Value;
     use proxmox_resource_scheduling::node::NodeStats;
     use proxmox_resource_scheduling::resource::ResourceStats;
+    use proxmox_resource_scheduling::scheduler::ScoredMigration;
     use proxmox_resource_scheduling::usage::Usage;
 
     use crate::bindings::resource_scheduling::{
-        resource::PveResource, usage::StartedResourceAggregator,
+        resource::{
+            CompactMigrationCandidate, PveResource, decompose_compact_migration_candidates,
+        },
+        usage::StartedResourceAggregator,
     };
 
     perlmod::declare_magic!(Box<Scheduler> : &Scheduler as "PVE::RS::ResourceScheduling::Static");
@@ -154,6 +158,56 @@ pub mod pve_rs_resource_scheduling_static {
         usage.remove_resource(sid);
     }
 
+    /// Method: Returns the load imbalance among the nodes.
+    ///
+    /// See [`proxmox_resource_scheduling::scheduler::Scheduler::node_imbalance`].
+    #[export]
+    pub fn calculate_node_imbalance(#[try_from_ref] this: &Scheduler) -> f64 {
+        let usage = this.inner.lock().unwrap();
+
+        usage
+            .to_scheduler::<StartedResourceAggregator>()
+            .node_imbalance()
+    }
+
+    /// Method: Scores the given migration `candidates` by the best node imbalance improvement with
+    /// exhaustive search.
+    ///
+    /// See [`proxmox_resource_scheduling::scheduler::Scheduler::score_best_balancing_migration_candidates`].
+    #[export]
+    pub fn score_best_balancing_migration_candidates(
+        #[try_from_ref] this: &Scheduler,
+        candidates: Vec<CompactMigrationCandidate>,
+        limit: usize,
+    ) -> Result<Vec<ScoredMigration>, Error> {
+        let usage = this.inner.lock().unwrap();
+
+        let candidates = decompose_compact_migration_candidates(&usage, candidates)?;
+
+        Ok(usage
+            .to_scheduler::<StartedResourceAggregator>()
+            .score_best_balancing_migration_candidates(candidates, limit))
+    }
+
+    /// Method: Scores the given migration `candidates` by the best node imbalance improvement with
+    /// the TOPSIS method.
+    ///
+    /// See [`proxmox_resource_scheduling::scheduler::Scheduler::score_best_balancing_migration_candidates_topsis`].
+    #[export]
+    pub fn score_best_balancing_migration_candidates_topsis(
+        #[try_from_ref] this: &Scheduler,
+        candidates: Vec<CompactMigrationCandidate>,
+        limit: usize,
+    ) -> Result<Vec<ScoredMigration>, Error> {
+        let usage = this.inner.lock().unwrap();
+
+        let candidates = decompose_compact_migration_candidates(&usage, candidates)?;
+
+        usage
+            .to_scheduler::<StartedResourceAggregator>()
+            .score_best_balancing_migration_candidates_topsis(&candidates, limit)
+    }
+
     /// Method: Scores nodes to start a service with the usage statistics `service_stats` on.
     ///
     /// See [`proxmox_resource_scheduling::scheduler::Scheduler::score_nodes_to_start_resource`].
diff --git a/pve-rs/src/bindings/resource_scheduling/resource.rs b/pve-rs/src/bindings/resource_scheduling/resource.rs
index 91d56b9..9186d5b 100644
--- a/pve-rs/src/bindings/resource_scheduling/resource.rs
+++ b/pve-rs/src/bindings/resource_scheduling/resource.rs
@@ -1,6 +1,8 @@
 use anyhow::{Error, bail};
-use proxmox_resource_scheduling::resource::{
-    Resource, ResourcePlacement, ResourceState, ResourceStats,
+use proxmox_resource_scheduling::{
+    resource::{Resource, ResourcePlacement, ResourceState, ResourceStats},
+    scheduler::{Migration, MigrationCandidate},
+    usage::Usage,
 };
 
 use serde::{Deserialize, Serialize};
@@ -42,3 +44,85 @@ impl<T: Into<ResourceStats>> TryFrom<PveResource<T>> for Resource {
         Ok(Resource::new(resource.stats.into(), state, placement))
     }
 }
+
+/// A compact representation of [`proxmox_resource_scheduling::scheduler::MigrationCandidate`].
+#[derive(Serialize, Deserialize)]
+pub struct CompactMigrationCandidate {
+    /// The identifier of the leading resource.
+    pub leader: String,
+    /// The resources which are part of the leading resource's bundle.
+    pub resources: Vec<String>,
+    /// The nodes, which are possible to migrate to for the resources.
+    pub nodes: Vec<String>,
+}
+
+/// Transforms a `Vec<CompactMigrationCandidate>` to a `Vec<MigrationCandidate>` with the cluster
+/// usage from `usage`.
+///
+/// This function fails for any of the following conditions for a [`CompactMigrationCandidate`]:
+///
+/// - the `leader` is not present in the cluster usage
+/// - the `leader` is non-stationary
+/// - any resource in `resources` is not present in the cluster usage
+/// - any resource in `resources` is non-stationary
+/// - any resource in `resources` is on another node than the `leader`
+pub(crate) fn decompose_compact_migration_candidates(
+    usage: &Usage,
+    compact_candidates: Vec<CompactMigrationCandidate>,
+) -> Result<Vec<MigrationCandidate>, Error> {
+    // The length of `compact_candidates` is at least a lower bound
+    let mut candidates = Vec::with_capacity(compact_candidates.len());
+
+    for candidate in compact_candidates.into_iter() {
+        let leader_sid = candidate.leader;
+        let leader = match usage.get_resource(&leader_sid) {
+            Some(resource) => resource,
+            _ => bail!("leader '{leader_sid}' is not present in the cluster usage"),
+        };
+        let leader_node = match leader.placement() {
+            ResourcePlacement::Stationary { current_node } => current_node,
+            _ => bail!("leader '{leader_sid}' is non-stationary"),
+        };
+
+        if !candidate.resources.contains(&leader_sid) {
+            bail!("leader '{leader_sid}' is not present in the resources list");
+        }
+
+        let mut resource_stats = Vec::with_capacity(candidate.resources.len());
+
+        for sid in candidate.resources.iter() {
+            let resource = match usage.get_resource(sid) {
+                Some(resource) => resource,
+                _ => bail!("resource '{sid}' is not present in the cluster usage"),
+            };
+
+            match resource.placement() {
+                ResourcePlacement::Stationary { current_node } => {
+                    if current_node != leader_node {
+                        bail!("resource '{sid}' is on other node than leader");
+                    }
+
+                    resource_stats.push(resource.stats());
+                }
+                _ => bail!("resource '{sid}' is non-stationary"),
+            }
+        }
+
+        let bundle_stats = resource_stats.into_iter().sum();
+
+        for target_node in candidate.nodes.into_iter() {
+            let migration = Migration {
+                sid: leader_sid.to_string(),
+                source_node: leader_node.to_string(),
+                target_node,
+            };
+
+            candidates.push(MigrationCandidate {
+                migration,
+                stats: bundle_stats,
+            });
+        }
+    }
+
+    Ok(candidates)
+}
diff --git a/pve-rs/src/bindings/resource_scheduling/usage.rs b/pve-rs/src/bindings/resource_scheduling/usage.rs
index 87b7e3e..48f6e84 100644
--- a/pve-rs/src/bindings/resource_scheduling/usage.rs
+++ b/pve-rs/src/bindings/resource_scheduling/usage.rs
@@ -4,6 +4,21 @@ use proxmox_resource_scheduling::{
     usage::{Usage, UsageAggregator},
 };
 
+/// The identity aggregator, which passes the node stats as-is.
+pub(crate) struct IdentityAggregator;
+
+impl UsageAggregator for IdentityAggregator {
+    fn aggregate(usage: &Usage) -> Vec<NodeUsage> {
+        usage
+            .nodes_iter()
+            .map(|(nodename, node)| NodeUsage {
+                name: nodename.to_string(),
+                stats: node.stats(),
+            })
+            .collect()
+    }
+}
+
 /// An aggregator, which adds any resource as a started resource.
 ///
 /// This aggregator is useful if the node base stats do not have any current usage.
-- 
2.47.3