[PATCH perl-rs v3 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>
Reviewed-by: Dominik Rusovac <d.rusovac@proxmox.com>
---
changes v2 -> v3:
- make node_stats closure variable mutable instead of shadowing it in
  the closure body again in StartedResourceAggregator::aggregate()
  (as suggested by @Dominik)

 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 3bbc464..c2f9b73 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::SDN \
 	  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..27ccf39
--- /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_owned())
+            .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.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 17a8d4d..d56a423 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(), |mut node_stats, sid| {
+                        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_owned(),
+                    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