[PATCH perl-rs v2 15/40] pve-rs: resource-scheduling: implement pve_dynamic bindings

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

The implementation is similar to pve_static, but extends the node and
resource stats with sampled runtime usage statistics, i.e., the actual
usage on the nodes and the actual usages of the resources.

In the case of users repeatedly calling score_nodes_to_start_resource()
and then adding them as starting resources with add_resource(), these
starting resources need to be accumulated on top of these nodes actual
current usages to prevent score_nodes_to_start_resource() to favor the
currently least loaded node(s) for all starting resources.

Signed-off-by: Daniel Kral <d.kral@proxmox.com>
---
changes v1 -> v2:
- move this patch one before 'expose auto rebalancing methods' as this
  is the same change order as done in pve-ha-manager, making it easier
  to separate the feature of using dynamic usage information and
  afterwards allowing rebalancing methods with static and dynamic usage
  information
- adapt patch message accordingly
- s/service/resource/ for any new struct and method as this is more
  consistent with the naming in the HA Manager and the name of the
  crate/module itself; can change this back if it's better in the other
  way, but as these are new API endpoints, I thought it's better to do
  it now than later

 pve-rs/Makefile                               |   1 +
 .../src/bindings/resource_scheduling/mod.rs   |   3 +
 .../resource_scheduling/pve_dynamic.rs        | 174 ++++++++++++++++++
 .../src/bindings/resource_scheduling/usage.rs |  33 ++++
 pve-rs/test/resource_scheduling.pl            |   1 +
 5 files changed, 212 insertions(+)
 create mode 100644 pve-rs/src/bindings/resource_scheduling/pve_dynamic.rs

diff --git a/pve-rs/Makefile b/pve-rs/Makefile
index 9faa735..f0212b7 100644
--- a/pve-rs/Makefile
+++ b/pve-rs/Makefile
@@ -30,6 +30,7 @@ PERLMOD_PACKAGES := \
 	  PVE::RS::OCI \
 	  PVE::RS::OpenId \
 	  PVE::RS::ResourceScheduling::Static \
+	  PVE::RS::ResourceScheduling::Dynamic \
 	  PVE::RS::SDN::Fabrics \
 	  PVE::RS::TFA
 
diff --git a/pve-rs/src/bindings/resource_scheduling/mod.rs b/pve-rs/src/bindings/resource_scheduling/mod.rs
index 9ce631c..87b4a03 100644
--- a/pve-rs/src/bindings/resource_scheduling/mod.rs
+++ b/pve-rs/src/bindings/resource_scheduling/mod.rs
@@ -5,3 +5,6 @@ mod usage;
 
 mod pve_static;
 pub use pve_static::pve_rs_resource_scheduling_static;
+
+mod pve_dynamic;
+pub use pve_dynamic::pve_rs_resource_scheduling_dynamic;
diff --git a/pve-rs/src/bindings/resource_scheduling/pve_dynamic.rs b/pve-rs/src/bindings/resource_scheduling/pve_dynamic.rs
new file mode 100644
index 0000000..5b4373e
--- /dev/null
+++ b/pve-rs/src/bindings/resource_scheduling/pve_dynamic.rs
@@ -0,0 +1,174 @@
+#[perlmod::package(name = "PVE::RS::ResourceScheduling::Dynamic", lib = "pve_rs")]
+pub mod pve_rs_resource_scheduling_dynamic {
+    //! The `PVE::RS::ResourceScheduling::Dynamic` package.
+    //!
+    //! Provides bindings for the dynamic resource scheduling module.
+    //!
+    //! See [`proxmox_resource_scheduling`].
+
+    use std::sync::Mutex;
+
+    use anyhow::Error;
+    use serde::{Deserialize, Serialize};
+
+    use perlmod::Value;
+    use proxmox_resource_scheduling::node::NodeStats;
+    use proxmox_resource_scheduling::resource::ResourceStats;
+    use proxmox_resource_scheduling::usage::Usage;
+
+    use crate::bindings::resource_scheduling::resource::PveResource;
+    use crate::bindings::resource_scheduling::usage::StartingAsStartedResourceAggregator;
+
+    perlmod::declare_magic!(Box<Scheduler> : &Scheduler as "PVE::RS::ResourceScheduling::Dynamic");
+
+    /// A scheduler instance contains the cluster usage.
+    pub struct Scheduler {
+        inner: Mutex<Usage>,
+    }
+
+    #[derive(Clone, Copy, Debug, Serialize, Deserialize)]
+    #[serde(rename_all = "kebab-case")]
+    /// Dynamic usage stats of a node.
+    pub struct DynamicNodeStats {
+        /// CPU utilization in CPU cores.
+        pub cpu: f64,
+        /// Total number of CPU cores.
+        pub maxcpu: usize,
+        /// Used memory in bytes.
+        pub mem: usize,
+        /// Total memory in bytes.
+        pub maxmem: usize,
+    }
+
+    impl From<DynamicNodeStats> for NodeStats {
+        fn from(value: DynamicNodeStats) -> Self {
+            Self {
+                cpu: value.cpu,
+                maxcpu: value.maxcpu,
+                mem: value.mem,
+                maxmem: value.maxmem,
+            }
+        }
+    }
+
+    #[derive(Clone, Copy, Debug, Serialize, Deserialize)]
+    #[serde(rename_all = "kebab-case")]
+    /// Dynamic usage stats of a resource.
+    pub struct DynamicResourceStats {
+        /// CPU utilization in CPU cores.
+        pub cpu: f64,
+        /// Number of assigned CPUs or CPU limit.
+        pub maxcpu: f64,
+        /// Used memory in bytes.
+        pub mem: usize,
+        /// Maximum assigned memory in bytes.
+        pub maxmem: usize,
+    }
+
+    impl From<DynamicResourceStats> for ResourceStats {
+        fn from(value: DynamicResourceStats) -> Self {
+            Self {
+                cpu: value.cpu,
+                maxcpu: value.maxcpu,
+                mem: value.mem,
+                maxmem: value.maxmem,
+            }
+        }
+    }
+
+    type DynamicResource = PveResource<DynamicResourceStats>;
+
+    /// Class method: Create a new [`Scheduler`] instance.
+    ///
+    /// See [`proxmox_resource_scheduling::usage::Usage::new`].
+    #[export(raw_return)]
+    pub fn new(#[raw] class: Value) -> Result<Value, Error> {
+        let inner = Usage::new();
+
+        Ok(perlmod::instantiate_magic!(
+            &class, MAGIC => Box::new(Scheduler { inner: Mutex::new(inner) })
+        ))
+    }
+
+    /// Method: Add a node with its basic CPU and memory info.
+    ///
+    /// See [`proxmox_resource_scheduling::usage::Usage::add_node`].
+    #[export]
+    pub fn add_node(
+        #[try_from_ref] this: &Scheduler,
+        nodename: String,
+        stats: DynamicNodeStats,
+    ) -> Result<(), Error> {
+        let mut usage = this.inner.lock().unwrap();
+
+        usage.add_node(nodename, stats.into())
+    }
+
+    /// Method: Remove a node from the scheduler.
+    ///
+    /// See [`proxmox_resource_scheduling::usage::Usage::remove_node`].
+    #[export]
+    pub fn remove_node(#[try_from_ref] this: &Scheduler, nodename: &str) {
+        let mut usage = this.inner.lock().unwrap();
+
+        usage.remove_node(nodename);
+    }
+
+    /// Method: Get a list of all the nodes in the scheduler.
+    #[export]
+    pub fn list_nodes(#[try_from_ref] this: &Scheduler) -> Vec<String> {
+        let usage = this.inner.lock().unwrap();
+
+        usage
+            .nodenames_iter()
+            .map(|nodename| nodename.to_string())
+            .collect()
+    }
+
+    /// Method: Check whether a node exists in the scheduler.
+    #[export]
+    pub fn contains_node(#[try_from_ref] this: &Scheduler, nodename: &str) -> bool {
+        let usage = this.inner.lock().unwrap();
+
+        usage.contains_node(nodename)
+    }
+
+    /// Method: Add `resource` with identifier `sid` to the scheduler.
+    ///
+    /// See [`proxmox_resource_scheduling::usage::Usage::add_resource`].
+    #[export]
+    pub fn add_resource(
+        #[try_from_ref] this: &Scheduler,
+        sid: String,
+        resource: DynamicResource,
+    ) -> Result<(), Error> {
+        let mut usage = this.inner.lock().unwrap();
+
+        usage.add_resource(sid, resource.try_into()?)
+    }
+
+    /// Method: Remove resource `sid` and its usage from all assigned nodes.
+    ///
+    /// See [`proxmox_resource_scheduling::usage::Usage::remove_resource`].
+    #[export]
+    fn remove_resource(#[try_from_ref] this: &Scheduler, sid: &str) {
+        let mut usage = this.inner.lock().unwrap();
+
+        usage.remove_resource(sid);
+    }
+
+    /// 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`].
+    #[export]
+    pub fn score_nodes_to_start_resource(
+        #[try_from_ref] this: &Scheduler,
+        resource_stats: DynamicResourceStats,
+    ) -> Result<Vec<(String, f64)>, Error> {
+        let usage = this.inner.lock().unwrap();
+
+        usage
+            .to_scheduler::<StartingAsStartedResourceAggregator>()
+            .score_nodes_to_start_resource(resource_stats)
+    }
+}
diff --git a/pve-rs/src/bindings/resource_scheduling/usage.rs b/pve-rs/src/bindings/resource_scheduling/usage.rs
index fc8b872..87b7e3e 100644
--- a/pve-rs/src/bindings/resource_scheduling/usage.rs
+++ b/pve-rs/src/bindings/resource_scheduling/usage.rs
@@ -1,4 +1,5 @@
 use proxmox_resource_scheduling::{
+    resource::ResourceState,
     scheduler::NodeUsage,
     usage::{Usage, UsageAggregator},
 };
@@ -31,3 +32,35 @@ impl UsageAggregator for StartedResourceAggregator {
             .collect()
     }
 }
+
+/// An aggregator, which uses the node base stats and adds any starting resources as already
+/// started resources to the node stats.
+///
+/// This aggregator is useful if starting resources should be considered in the scheduler.
+pub(crate) struct StartingAsStartedResourceAggregator;
+
+impl UsageAggregator for StartingAsStartedResourceAggregator {
+    fn aggregate(usage: &Usage) -> Vec<NodeUsage> {
+        usage
+            .nodes_iter()
+            .map(|(nodename, node)| {
+                let stats = node.resources_iter().fold(node.stats(), |node_stats, sid| {
+                    let mut node_stats = node_stats;
+
+                    if let Some(resource) = usage.get_resource(sid)
+                        && resource.state() == ResourceState::Starting
+                    {
+                        node_stats.add_started_resource(&resource.stats());
+                    }
+
+                    node_stats
+                });
+
+                NodeUsage {
+                    name: nodename.to_string(),
+                    stats,
+                }
+            })
+            .collect()
+    }
+}
diff --git a/pve-rs/test/resource_scheduling.pl b/pve-rs/test/resource_scheduling.pl
index a332269..3775242 100755
--- a/pve-rs/test/resource_scheduling.pl
+++ b/pve-rs/test/resource_scheduling.pl
@@ -6,6 +6,7 @@ use warnings;
 use Test::More;
 
 use PVE::RS::ResourceScheduling::Static;
+use PVE::RS::ResourceScheduling::Dynamic;
 
 my sub score_nodes {
     my ($static, $service) = @_;
-- 
2.47.3