[PATCH proxmox v2 05/40] resource-scheduling: implement generic cluster usage implementation

Daniel Kral
2026-03-24T18:32:51Z

This is a more generic version of the `Usage` implementation from the
pve_static bindings in the pve_rs repository.

As the upcoming load balancing scheduler actions and dynamic resource
scheduler will need more information about each resource, this further
improves on the state tracking of each resource:

In this implementation, a resource is composed of its usage statistics
and its two essential states: the running state and the node placement.
The non_exhaustive attribute ensures that usages need to construct the
a Resource instance through its API.

Users can repeatedly use the current state of Usage to make scheduling
decisions with the to_scheduler() method. This method takes an
implementation of UsageAggregator, which dictates how the usage
information is represented to the Scheduler.

Signed-off-by: Daniel Kral <d.kral@proxmox.com>
---
changes v1 -> v2:
- new!

This patch is added to move the handling of specific usage stats and
their (de)serialization to the pve-rs bindings and have the general
functionality in this crate.

proxmox-resource-scheduling/src/lib.rs | 1 +
proxmox-resource-scheduling/src/node.rs | 40 +++++
proxmox-resource-scheduling/src/resource.rs | 119 +++++++++++++
proxmox-resource-scheduling/src/usage.rs | 183 ++++++++++++++++++++
proxmox-resource-scheduling/tests/usage.rs | 153 ++++++++++++++++
5 files changed, 496 insertions(+)
create mode 100644 proxmox-resource-scheduling/src/usage.rs
create mode 100644 proxmox-resource-scheduling/tests/usage.rs

diff --git a/proxmox-resource-scheduling/src/lib.rs b/proxmox-resource-scheduling/src/lib.rs
index 12b743fe..99ca16d8 100644
--- a/proxmox-resource-scheduling/src/lib.rs
+++ b/proxmox-resource-scheduling/src/lib.rs
@@ -3,6 +3,7 @@ pub mod topsis;

pub mod node;
pub mod resource;
+pub mod usage;

pub mod scheduler;

diff --git a/proxmox-resource-scheduling/src/node.rs b/proxmox-resource-scheduling/src/node.rs
index e6227eda..be462782 100644
--- a/proxmox-resource-scheduling/src/node.rs
+++ b/proxmox-resource-scheduling/src/node.rs
@@ -1,3 +1,5 @@
+use std::collections::HashSet;
+
use crate::resource::ResourceStats;

/// Usage statistics of a node.
@@ -37,3 +39,41 @@ impl NodeStats {
self.mem as f64 / self.maxmem as f64
}
}
+
+/// A node in the cluster context.
+#[derive(Clone, Debug)]
+pub struct Node {
+ /// Base stats of the node.
+ stats: NodeStats,
+ /// The identifiers of the resources assigned to the node.
+ resources: HashSet<String>,
+}
+
+impl Node {
+ pub fn new(stats: NodeStats) -> Self {
+ Self {
+ stats,
+ resources: HashSet::new(),
+ }
+ }
+
+ pub fn add_resource(&mut self, sid: &str) -> bool {
+ self.resources.insert(sid.to_string())
+ }
+
+ pub fn remove_resource(&mut self, sid: &str) -> bool {
+ self.resources.remove(sid)
+ }
+
+ pub fn stats(&self) -> NodeStats {
+ self.stats
+ }
+
+ pub fn resources_iter(&self) -> impl Iterator<Item = &String> {
+ self.resources.iter()
+ }
+
+ pub fn contains_resource(&self, sid: &str) -> bool {
+ self.resources.contains(sid)
+ }
+}
diff --git a/proxmox-resource-scheduling/src/resource.rs b/proxmox-resource-scheduling/src/resource.rs
index 1eb9d15e..2aa16a51 100644
--- a/proxmox-resource-scheduling/src/resource.rs
+++ b/proxmox-resource-scheduling/src/resource.rs
@@ -1,5 +1,7 @@
use std::{iter::Sum, ops::Add};

+use anyhow::{bail, Error};
+
/// Usage statistics for a resource.
#[derive(Copy, Clone, PartialEq, PartialOrd, Debug, Default)]
pub struct ResourceStats {
@@ -31,3 +33,120 @@ impl Sum for ResourceStats {
iter.fold(Self::default(), |a, b| a + b)
}
}
+
+/// Execution state of a resource.
+#[derive(Copy, Clone, PartialEq, Eq, Debug)]
+#[non_exhaustive]
+pub enum ResourceState {
+ /// The resource is stopped.
+ Stopped,
+ /// The resource is scheduled to start.
+ Starting,
+ /// The resource is started and currently running.
+ Started,
+}
+
+/// Placement of a resource.
+#[derive(Clone, PartialEq, Eq, Debug)]
+#[non_exhaustive]
+pub enum ResourcePlacement {
+ /// The resource is on `current_node`.
+ Stationary { current_node: String },
+ /// The resource is being moved from `current_node` to `target_node`.
+ Moving {
+ current_node: String,
+ target_node: String,
+ },
+}
+
+impl ResourcePlacement {
+ fn nodenames(&self) -> Vec<&str> {
+ match self {
+ ResourcePlacement::Stationary { current_node } => vec![&current_node],
+ ResourcePlacement::Moving {
+ current_node,
+ target_node,
+ } => vec![&current_node, &target_node],
+ }
+ }
+}
+
+/// A resource in the cluster context.
+#[derive(Clone, Debug)]
+#[non_exhaustive]
+pub struct Resource {
+ /// The usage statistics of the resource.
+ stats: ResourceStats,
+ /// The execution state of the resource.
+ state: ResourceState,
+ /// The placement of the resource.
+ placement: ResourcePlacement,
+}
+
+impl Resource {
+ pub fn new(stats: ResourceStats, state: ResourceState, placement: ResourcePlacement) -> Self {
+ Self {
+ stats,
+ state,
+ placement,
+ }
+ }
+
+ /// Put the resource into a moving state with `target_node`.
+ ///
+ /// This method fails if the resource is already moving.
+ pub fn moving_to(&mut self, target_node: String) -> Result<(), Error> {
+ match &self.placement {
+ ResourcePlacement::Stationary { current_node } => {
+ self.placement = ResourcePlacement::Moving {
+ current_node: current_node.to_string(),
+ target_node,
+ };
+ }
+ ResourcePlacement::Moving { .. } => bail!("resource is already moving"),
+ };
+
+ Ok(())
+ }
+
+ /// Handles the external removal of a node.
+ ///
+ /// Returns whether the resource does not have any node left.
+ pub fn remove_node(&mut self, nodename: &str) -> bool {
+ match &self.placement {
+ ResourcePlacement::Stationary { current_node } => current_node == nodename,
+ ResourcePlacement::Moving {
+ current_node,
+ target_node,
+ } => {
+ if current_node == nodename {
+ self.placement = ResourcePlacement::Stationary {
+ current_node: target_node.to_string(),
+ };
+ } else if target_node == nodename {
+ self.placement = ResourcePlacement::Stationary {
+ current_node: current_node.to_string(),
+ };
+ }
+
+ false
+ }
+ }
+ }
+
+ pub fn state(&self) -> ResourceState {
+ self.state
+ }
+
+ pub fn stats(&self) -> ResourceStats {
+ self.stats
+ }
+
+ pub fn placement(&self) -> &ResourcePlacement {
+ &self.placement
+ }
+
+ pub fn nodenames(&self) -> Vec<&str> {
+ self.placement.nodenames()
+ }
+}
diff --git a/proxmox-resource-scheduling/src/usage.rs b/proxmox-resource-scheduling/src/usage.rs
new file mode 100644
index 00000000..78ccc453
--- /dev/null
+++ b/proxmox-resource-scheduling/src/usage.rs
@@ -0,0 +1,183 @@
+use anyhow::{bail, Error};
+
+use std::collections::HashMap;
+
+use crate::{
+ node::{Node, NodeStats},
+ resource::{Resource, ResourcePlacement, ResourceState, ResourceStats},
+ scheduler::{NodeUsage, Scheduler},
+};
+
+/// The state of the usage in the cluster.
+///
+/// The cluster usage represents the current state of the assignments between nodes and resources
+/// and their usage statistics. A resource can be placed on these nodes according to their
+/// placement state. See [`crate::resource::Resource`] for more information.
+///
+/// The cluster usage state can be used to build a current state for the [`Scheduler`].
+#[derive(Default)]
+pub struct Usage {
+ nodes: HashMap<String, Node>,
+ resources: HashMap<String, Resource>,
+}
+
+/// An aggregator for the [`Usage`] maps the cluster usage to node usage statistics that are
+/// relevant for the scheduler.
+pub trait UsageAggregator {
+ fn aggregate(usage: &Usage) -> Vec<NodeUsage>;
+}
+
+impl Usage {
+ /// Instantiate an empty cluster usage.
+ pub fn new() -> Self {
+ Self::default()
+ }
+
+ /// Add a node to the cluster usage.
+ ///
+ /// This method fails if a node with the same `nodename` already exists.
+ pub fn add_node(&mut self, nodename: String, stats: NodeStats) -> Result<(), Error> {
+ if self.nodes.contains_key(&nodename) {
+ bail!("node '{}' already exists", nodename);
+ }
+
+ self.nodes.insert(nodename, Node::new(stats));
+
+ Ok(())
+ }
+
+ /// Remove a node from the cluster usage.
+ pub fn remove_node(&mut self, nodename: &str) {
+ if let Some(node) = self.nodes.remove(nodename) {
+ node.resources_iter().for_each(|sid| {
+ if let Some(resource) = self.resources.get_mut(sid)
+ && resource.remove_node(nodename)
+ {
+ self.resources.remove(sid);
+ }
+ });
+ }
+ }
+
+ /// Returns a reference to the [`Node`] with the identifier `nodename`.
+ pub fn get_node(&self, nodename: &str) -> Option<&Node> {
+ self.nodes.get(nodename)
+ }
+
+ /// Returns an iterator for the cluster usage's nodes.
+ pub fn nodes_iter(&self) -> impl Iterator<Item = (&String, &Node)> {
+ self.nodes.iter()
+ }
+
+ /// Returns an iterator for the cluster usage's nodes.
+ pub fn nodenames_iter(&self) -> impl Iterator<Item = &String> {
+ self.nodes.keys()
+ }
+
+ /// Returns whether the node with the identifier `nodename` is present in the cluster usage.
+ pub fn contains_node(&self, nodename: &str) -> bool {
+ self.nodes.contains_key(nodename)
+ }
+
+ fn add_resource_to_nodes(&mut self, sid: &str, nodenames: Vec<&str>) -> Result<(), Error> {
+ if nodenames
+ .iter()
+ .any(|nodename| !self.nodes.contains_key(*nodename))
+ {
+ bail!("resource nodes do not exist");
+ }
+
+ nodenames.iter().for_each(|nodename| {
+ if let Some(node) = self.nodes.get_mut(*nodename) {
+ node.add_resource(sid);
+ }
+ });
+
+ Ok(())
+ }
+
+ fn remove_resource_from_nodes(&mut self, sid: &str, nodenames: &[&str]) {
+ nodenames.iter().for_each(|nodename| {
+ if let Some(node) = self.nodes.get_mut(*nodename) {
+ node.remove_resource(sid);
+ }
+ });
+ }
+
+ /// Add `resource` with identifier `sid` to cluster usage.
+ ///
+ /// This method fails if a resource with the same `sid` already exists or the resource's nodes
+ /// do not exist in the cluster usage.
+ pub fn add_resource(&mut self, sid: String, resource: Resource) -> Result<(), Error> {
+ if self.resources.contains_key(&sid) {
+ bail!("resource '{}' already exists", sid);
+ }
+
+ self.add_resource_to_nodes(&sid, resource.nodenames())?;
+
+ self.resources.insert(sid.to_string(), resource);
+
+ Ok(())
+ }
+
+ /// Add `stats` from resource with identifier `sid` to node `nodename` in cluster usage.
+ ///
+ /// For the first call, the resource is assumed to be started and stationary on the given node.
+ /// If there was no intermediate call to remove the resource, the second call will assume that
+ /// the given node is the target node and the resource is being moved there. The second call
+ /// will ignore the value of `stats`.
+ #[deprecated = "only for backwards compatibility, use add_resource(...) instead"]
+ pub fn add_resource_usage_to_node(
+ &mut self,
+ nodename: &str,
+ sid: &str,
+ stats: ResourceStats,
+ ) -> Result<(), Error> {
+ if let Some(resource) = self.resources.get_mut(sid) {
+ resource.moving_to(nodename.to_string())?;
+
+ self.add_resource_to_nodes(sid, vec![nodename])
+ } else {
+ let placement = ResourcePlacement::Stationary {
+ current_node: nodename.to_string(),
+ };
+ let resource = Resource::new(stats, ResourceState::Started, placement);
+
+ self.add_resource(sid.to_string(), resource)
+ }
+ }
+
+ /// Remove resource with identifier `sid` from cluster usage.
+ pub fn remove_resource(&mut self, sid: &str) {
+ if let Some(resource) = self.resources.remove(sid) {
+ match resource.placement() {
+ ResourcePlacement::Stationary { current_node } => {
+ self.remove_resource_from_nodes(sid, &[current_node]);
+ }
+ ResourcePlacement::Moving {
+ current_node,
+ target_node,
+ } => {
+ self.remove_resource_from_nodes(sid, &[current_node, target_node]);
+ }
+ }
+ }
+ }
+
+ /// Returns a reference to the [`Resource`] with the identifier `sid`.
+ pub fn get_resource(&self, sid: &str) -> Option<&Resource> {
+ self.resources.get(sid)
+ }
+
+ /// Returns an iterator for the cluster usage's resources.
+ pub fn resources_iter(&self) -> impl Iterator<Item = (&String, &Resource)> {
+ self.resources.iter()
+ }
+
+ /// Use the current cluster usage as a base for a scheduling action.
+ pub fn to_scheduler<F: UsageAggregator>(&self) -> Scheduler {
+ let node_usages = F::aggregate(self);
+
+ Scheduler::from_nodes(node_usages)
+ }
+}
diff --git a/proxmox-resource-scheduling/tests/usage.rs b/proxmox-resource-scheduling/tests/usage.rs
new file mode 100644
index 00000000..eb00d2c6
--- /dev/null
+++ b/proxmox-resource-scheduling/tests/usage.rs
@@ -0,0 +1,153 @@
+use anyhow::{bail, Error};
+use proxmox_resource_scheduling::{
+ node::NodeStats,
+ resource::{Resource, ResourcePlacement, ResourceState, ResourceStats},
+ usage::Usage,
+};
+
+#[test]
+fn test_no_duplicate_nodes() -> Result<(), Error> {
+ let mut usage = Usage::new();
+
+ usage.add_node("node1".to_string(), NodeStats::default())?;
+
+ match usage.add_node("node1".to_string(), NodeStats::default()) {
+ Ok(_) => bail!("cluster usage does allow duplicate node entries"),
+ Err(_) => Ok(()),
+ }
+}
+
+#[test]
+fn test_no_duplicate_resources() -> Result<(), Error> {
+ let mut usage = Usage::new();
+
+ usage.add_node("node1".to_string(), NodeStats::default())?;
+
+ let placement = ResourcePlacement::Stationary {
+ current_node: "node1".to_string(),
+ };
+ let resource = Resource::new(ResourceStats::default(), ResourceState::Stopped, placement);
+
+ usage.add_resource("vm:101".to_string(), resource.clone())?;
+
+ match usage.add_resource("vm:101".to_string(), resource) {
+ Ok(_) => bail!("cluster usage does allow duplicate resource entries"),
+ Err(_) => Ok(()),
+ }
+}
+
+#[test]
+#[allow(deprecated)]
+fn test_add_resource_usage_to_node() -> Result<(), Error> {
+ let mut usage = Usage::new();
+
+ usage.add_node("node1".to_string(), NodeStats::default())?;
+ usage.add_node("node2".to_string(), NodeStats::default())?;
+ usage.add_node("node3".to_string(), NodeStats::default())?;
+
+ usage.add_resource_usage_to_node("node1", "vm:101", ResourceStats::default())?;
+ usage.add_resource_usage_to_node("node2", "vm:101", ResourceStats::default())?;
+
+ if usage
+ .add_resource_usage_to_node("node3", "vm:101", ResourceStats::default())
+ .is_ok()
+ {
+ bail!("add_resource_usage_to_node() allows adding resource to more than two nodes");
+ }
+
+ Ok(())
+}
+
+#[test]
+fn test_add_remove_stationary_resource() -> Result<(), Error> {
+ let mut usage = Usage::new();
+
+ let (sid, nodename) = ("vm:101", "node1");
+
+ usage.add_node(nodename.to_string(), NodeStats::default())?;
+
+ let placement = ResourcePlacement::Stationary {
+ current_node: nodename.to_string(),
+ };
+ let resource = Resource::new(ResourceStats::default(), ResourceState::Stopped, placement);
+
+ usage.add_resource(sid.to_string(), resource)?;
+
+ match (usage.get_resource(sid), usage.get_node(nodename)) {
+ (Some(_), Some(node)) => {
+ if !node.contains_resource(sid) {
+ bail!("resource '{sid}' was not added to node '{nodename}'");
+ }
+ }
+ _ => bail!("resource '{sid}' or node '{nodename}' were not added"),
+ }
+
+ usage.remove_resource(sid);
+
+ match (usage.get_resource(sid), usage.get_node(nodename)) {
+ (None, Some(node)) => {
+ if node.contains_resource(sid) {
+ bail!("resource '{sid}' was not removed from node '{nodename}'");
+ }
+ }
+ _ => bail!("resource '{sid}' was not removed"),
+ }
+
+ Ok(())
+}
+
+#[test]
+fn test_add_remove_moving_resource() -> Result<(), Error> {
+ let mut usage = Usage::new();
+
+ let (sid, current_nodename, target_nodename) = ("vm:101", "node1", "node2");
+
+ usage.add_node(current_nodename.to_string(), NodeStats::default())?;
+ usage.add_node(target_nodename.to_string(), NodeStats::default())?;
+
+ let placement = ResourcePlacement::Moving {
+ current_node: current_nodename.to_string(),
+ target_node: target_nodename.to_string(),
+ };
+ let resource = Resource::new(ResourceStats::default(), ResourceState::Stopped, placement);
+
+ usage.add_resource(sid.to_string(), resource)?;
+
+ match (
+ usage.get_resource(sid),
+ usage.get_node(current_nodename),
+ usage.get_node(target_nodename),
+ ) {
+ (Some(_), Some(current_node), Some(target_node)) => {
+ if !current_node.contains_resource("vm:101") {
+ bail!("resource '{sid}' was not added to current node '{current_nodename}'");
+ }
+
+ if !target_node.contains_resource("vm:101") {
+ bail!("resource '{sid}' was not added to target node '{target_nodename}'");
+ }
+ }
+ _ => bail!("resource '{sid}' or nodes were not added"),
+ }
+
+ usage.remove_resource(sid);
+
+ match (
+ usage.get_resource(sid),
+ usage.get_node(current_nodename),
+ usage.get_node(target_nodename),
+ ) {
+ (None, Some(current_node), Some(target_node)) => {
+ if current_node.contains_resource(sid) {
+ bail!("resource '{sid}' was not removed from current node '{current_nodename}'");
+ }
+
+ if target_node.contains_resource(sid) {
+ bail!("resource '{sid}' was not removed from target node '{target_nodename}'");
+ }
+ }
+ _ => bail!("resource '{sid}' was not removed"),
+ }
+
+ Ok(())
+}
--
2.47.3