From 329876da0b1abc96cb2a22d7c9ceaade4ac520dd Mon Sep 17 00:00:00 2001 From: Yoann Lamouroux Date: Mon, 12 May 2025 14:57:11 +0200 Subject: [PATCH 1/7] feat: bring in the `balance` logic, does nothing just yet --- requirements.txt | 1 + src/pvecontrol/__init__.py | 1 + src/pvecontrol/actions/__init__.py | 2 +- src/pvecontrol/actions/cluster_balance.py | 151 ++++++++++++++++++++++ 4 files changed, 154 insertions(+), 1 deletion(-) create mode 100644 src/pvecontrol/actions/cluster_balance.py diff --git a/requirements.txt b/requirements.txt index 4b3955ec..2903658f 100644 --- a/requirements.txt +++ b/requirements.txt @@ -12,3 +12,4 @@ requests requests_toolbelt openssh_wrapper paramiko +ortools \ No newline at end of file diff --git a/src/pvecontrol/__init__.py b/src/pvecontrol/__init__.py index eb02a514..16db701b 100644 --- a/src/pvecontrol/__init__.py +++ b/src/pvecontrol/__init__.py @@ -136,6 +136,7 @@ def pvecontrol(ctx, debug, output, cluster): pvecontrol.add_command(cmd=actions.cluster.status, name="status") pvecontrol.add_command(cmd=actions.cluster.sanitycheck, name="sanitycheck") +pvecontrol.add_command(cmd=actions.cluster_balance.balance, name="balance") pvecontrol.add_command(cmd=actions.node.root, name="node") pvecontrol.add_command(cmd=actions.storage.root, name="storage") pvecontrol.add_command(cmd=actions.task.root, name="task") diff --git a/src/pvecontrol/actions/__init__.py b/src/pvecontrol/actions/__init__.py index b71e17a1..5187f72a 100644 --- a/src/pvecontrol/actions/__init__.py +++ b/src/pvecontrol/actions/__init__.py @@ -1 +1 @@ -from . import cluster, node, storage, task, vm +from . import cluster, node, storage, task, vm, cluster_balance diff --git a/src/pvecontrol/actions/cluster_balance.py b/src/pvecontrol/actions/cluster_balance.py new file mode 100644 index 00000000..d823ac4f --- /dev/null +++ b/src/pvecontrol/actions/cluster_balance.py @@ -0,0 +1,151 @@ +from itertools import chain +import click + +from pvecontrol.models.cluster import PVECluster +from pvecontrol.models.node import PVENode +from pvecontrol.models.vm import PVEVm +from ortools.sat.python import cp_model +# import numpy as np + + +@click.command() +@click.option( + "-a", "--anti-affinity", default="", help="Balance machine with those comma-separated prefixes across nodes" +) +@click.option("-p", "--movement-penalty", default=0, type=int, help="Higher values minimize VM movement") +@click.option("-d", "--dry-run", is_flag=True, help="Only print the `vm migrate` commands, doesn't take action") +@click.pass_context +def balance(ctx, anti_affinity, movement_penalty, dry_run): + """Balances VMs across nodes in the cluster""" + overcommit_factor = 1.2 + proxmox: PVECluster = PVECluster.create_from_config(ctx.obj["args"].cluster) + nodes: list[PVENode] = proxmox.nodes + vms: list[PVEVm] = list(chain.from_iterable(node.vms for node in nodes)) + vms_count = len(vms) + nodes_count = len(nodes) + vm_params = [{"cpu": vm.cpus, "memory": vm.maxmem} for vm in vms] + node_caps = list({"cpu": node.maxcpu, "memory": node.maxmem} for node in nodes) + vm_groups = { + aag: list(map(lambda i: i[0], filter(lambda x: aag in x[1].name, enumerate(vms)))) + for aag in anti_affinity.split(",") + } + + node_map = {vm.name: node_idx for node_idx, node in enumerate(nodes) for vm in node.vms} + initial_locations = {i: node_map.get(vm.name) for i, vm in enumerate(vms) if vm.name in node_map} + + model = cp_model.CpModel() + vm_to_nodes = { + (vm_index, node_index): model.NewBoolVar(f"vm_{vm_index}_to_node_{node_index}") + for vm_index in range(vms_count) + for node_index in range(nodes_count) + } + + for vm_index in range(vms_count): + model.Add(sum(vm_to_nodes[(vm_index, node_index)] for node_index in range(nodes_count)) == 1) + + for node_index in range(nodes_count): + overcomitted_cpu_cap = int(node_caps[node_index]["cpu"] * overcommit_factor) + resources = {"cpu": overcomitted_cpu_cap, "memory": node_caps[node_index]["memory"]} + for k, v in resources.items(): + model.Add( + sum(vm_params[vm_index][k] * vm_to_nodes[(vm_index, node_index)] for vm_index in range(vms_count)) <= v + ) + + obj_terms = [] + for gid, vm_indices in vm_groups.items(): + if not vm_indices: continue + + max_vms_in_node = model.NewIntVar(0, len(vm_indices), f"max_vms_from_group_{gid}") + + group_vms_per_node = {} + for node_index in range(nodes_count): + group_vms_per_node[node_index] = model.NewIntVar(0, len(vm_indices), f"group_{gid}_vms_in_node_{node_index}") + model.Add(group_vms_per_node[node_index] == sum(vm_to_nodes[(vm_index, node_index)] for vm_index in vm_indices)) + model.Add(max_vms_in_node >= group_vms_per_node[node_index]) + + obj_terms.append(max_vms_in_node * len(vm_indices) * 100) + for node_index_start in range(nodes_count): + for node_index_remain in range(node_index_start+1, nodes_count): + diff = model.NewIntVar(0, len(vm_indices), f"diff_group_{gid}_nodes_{node_index_start}_{node_index_remain}") + model.AddAbsEquality(diff, group_vms_per_node[node_index_start] - group_vms_per_node[node_index_remain]) + obj_terms.append(diff * 10) + + vm_moved = {} + for vm_index in range(vms_count): + if node_index in initial_locations: + vm_moved[vm_index] = model.NewBoolVar(f"vm_{vm_index}_moved") + initial_node = initial_locations[vm_index] + model.Add(vm_to_nodes[(vm_index, initial_node)] == 1).OnlyEnforceIf(vm_moved[vm_index].Not()) + model.Add(vm_to_nodes[(vm_index, initial_node)] == 0).OnlyEnforceIf(vm_moved[vm_index]) + obj_terms.append(vm_moved[vm_index] * movement_penalty) + + resource_per_node = {} + max_resource = {} + resources = { + 'cpu': { + 'max_cap': int(max(cap['cpu'] for cap in node_caps) * overcommit_factor), + 'node_cap': lambda node_index: int(node_caps[node_index]['cpu'] * overcommit_factor) + }, + 'memory': { + 'max_cap': max(cap['memory'] for cap in node_caps), + 'node_cap': lambda node_index: node_caps[node_index]['memory'] + } + } + for rtype, rvalue in resources.items(): + resource_per_node[rtype] = {} + for node_index in range(nodes_count): + node_var = model.NewIntVar(0, rvalue['node_cap'](node_index), f"{rtype}_in_node_{node_index}") + resource_per_node[rtype][node_index] = node_var + model.Add(node_var == sum( + vm_params[vm_index][rtype] * vm_to_nodes[(vm_index, node_index)] for vm_index in range(vms_count))) + + max_resource[rtype] = model.NewIntVar(0, rvalue['max_cap'], f"max_{rtype}") + + max_cpu = max_resource['cpu'] + max_memory = max_resource['memory'] + cpu_per_node = resource_per_node['cpu'] + memory_per_node = resource_per_node['memory'] + + for node_index in range(nodes_count): + model.Add(max_cpu >= cpu_per_node[node_index]) + model.Add(max_memory >= memory_per_node[node_index]) + + obj_terms.append(max_cpu) + obj_terms.append(max_memory) + + model.Minimize(sum(obj_terms)) + solver = cp_model.CpSolver() + solver.parameters.max_time_in_seconds = 60.0 + status = solver.Solve(model=model) + if status == cp_model.OPTIMAL or status == cp_model.FEASIBLE: + assignments = {} + moves = [] + + for vm_index in range(vms_count): + for node_index in range(nodes_count): + if solver.Value(vm_to_nodes[(vm_index, node_index)]) == 1: + assignments[vm_index] = node_index + + if vm_index in initial_locations and initial_locations[vm_index] != node_index: + moves.append((vm_index, initial_locations[vm_index], node_index)) + # group_distributions = {} + # group_variances = {} + # for gid, vm_indices in vm_groups.items(): + # if not vm_indices: continue + + # group_dist = [0] * nodes_count + # for it in vm_indices: + # group_dist[assignments[it]] += 1 + + # group_distributions[gid] = group_dist + # group_variances[gid] = np.var(group_dist) + + print(moves) + else: + print('failed') + # print(vm_params) + # print(initial_locations) + # # metrics = proxmox.metrics + # print(anti_affinity) + # print(movement_penalty) + # print(dry_run) From 773b559b3bdb3ac6ec9e14716001b4df46862898 Mon Sep 17 00:00:00 2001 From: Yoann Lamouroux Date: Mon, 12 May 2025 15:30:24 +0200 Subject: [PATCH 2/7] black reformatting --- src/pvecontrol/actions/cluster_balance.py | 78 +++++++++++++---------- 1 file changed, 45 insertions(+), 33 deletions(-) diff --git a/src/pvecontrol/actions/cluster_balance.py b/src/pvecontrol/actions/cluster_balance.py index d823ac4f..7c58b429 100644 --- a/src/pvecontrol/actions/cluster_balance.py +++ b/src/pvecontrol/actions/cluster_balance.py @@ -5,6 +5,7 @@ from pvecontrol.models.node import PVENode from pvecontrol.models.vm import PVEVm from ortools.sat.python import cp_model + # import numpy as np @@ -50,26 +51,33 @@ def balance(ctx, anti_affinity, movement_penalty, dry_run): model.Add( sum(vm_params[vm_index][k] * vm_to_nodes[(vm_index, node_index)] for vm_index in range(vms_count)) <= v ) - + obj_terms = [] for gid, vm_indices in vm_groups.items(): - if not vm_indices: continue + if not vm_indices: + continue max_vms_in_node = model.NewIntVar(0, len(vm_indices), f"max_vms_from_group_{gid}") group_vms_per_node = {} for node_index in range(nodes_count): - group_vms_per_node[node_index] = model.NewIntVar(0, len(vm_indices), f"group_{gid}_vms_in_node_{node_index}") - model.Add(group_vms_per_node[node_index] == sum(vm_to_nodes[(vm_index, node_index)] for vm_index in vm_indices)) + group_vms_per_node[node_index] = model.NewIntVar( + 0, len(vm_indices), f"group_{gid}_vms_in_node_{node_index}" + ) + model.Add( + group_vms_per_node[node_index] == sum(vm_to_nodes[(vm_index, node_index)] for vm_index in vm_indices) + ) model.Add(max_vms_in_node >= group_vms_per_node[node_index]) obj_terms.append(max_vms_in_node * len(vm_indices) * 100) for node_index_start in range(nodes_count): - for node_index_remain in range(node_index_start+1, nodes_count): - diff = model.NewIntVar(0, len(vm_indices), f"diff_group_{gid}_nodes_{node_index_start}_{node_index_remain}") + for node_index_remain in range(node_index_start + 1, nodes_count): + diff = model.NewIntVar( + 0, len(vm_indices), f"diff_group_{gid}_nodes_{node_index_start}_{node_index_remain}" + ) model.AddAbsEquality(diff, group_vms_per_node[node_index_start] - group_vms_per_node[node_index_remain]) obj_terms.append(diff * 10) - + vm_moved = {} for vm_index in range(vms_count): if node_index in initial_locations: @@ -78,38 +86,42 @@ def balance(ctx, anti_affinity, movement_penalty, dry_run): model.Add(vm_to_nodes[(vm_index, initial_node)] == 1).OnlyEnforceIf(vm_moved[vm_index].Not()) model.Add(vm_to_nodes[(vm_index, initial_node)] == 0).OnlyEnforceIf(vm_moved[vm_index]) obj_terms.append(vm_moved[vm_index] * movement_penalty) - + resource_per_node = {} max_resource = {} resources = { - 'cpu': { - 'max_cap': int(max(cap['cpu'] for cap in node_caps) * overcommit_factor), - 'node_cap': lambda node_index: int(node_caps[node_index]['cpu'] * overcommit_factor) + "cpu": { + "max_cap": int(max(cap["cpu"] for cap in node_caps) * overcommit_factor), + "node_cap": lambda node_index: int(node_caps[node_index]["cpu"] * overcommit_factor), + }, + "memory": { + "max_cap": max(cap["memory"] for cap in node_caps), + "node_cap": lambda node_index: node_caps[node_index]["memory"], }, - 'memory': { - 'max_cap': max(cap['memory'] for cap in node_caps), - 'node_cap': lambda node_index: node_caps[node_index]['memory'] - } } for rtype, rvalue in resources.items(): resource_per_node[rtype] = {} for node_index in range(nodes_count): - node_var = model.NewIntVar(0, rvalue['node_cap'](node_index), f"{rtype}_in_node_{node_index}") + node_var = model.NewIntVar(0, rvalue["node_cap"](node_index), f"{rtype}_in_node_{node_index}") resource_per_node[rtype][node_index] = node_var - model.Add(node_var == sum( - vm_params[vm_index][rtype] * vm_to_nodes[(vm_index, node_index)] for vm_index in range(vms_count))) - - max_resource[rtype] = model.NewIntVar(0, rvalue['max_cap'], f"max_{rtype}") - - max_cpu = max_resource['cpu'] - max_memory = max_resource['memory'] - cpu_per_node = resource_per_node['cpu'] - memory_per_node = resource_per_node['memory'] + model.Add( + node_var + == sum( + vm_params[vm_index][rtype] * vm_to_nodes[(vm_index, node_index)] for vm_index in range(vms_count) + ) + ) + + max_resource[rtype] = model.NewIntVar(0, rvalue["max_cap"], f"max_{rtype}") + + max_cpu = max_resource["cpu"] + max_memory = max_resource["memory"] + cpu_per_node = resource_per_node["cpu"] + memory_per_node = resource_per_node["memory"] for node_index in range(nodes_count): model.Add(max_cpu >= cpu_per_node[node_index]) model.Add(max_memory >= memory_per_node[node_index]) - + obj_terms.append(max_cpu) obj_terms.append(max_memory) @@ -120,29 +132,29 @@ def balance(ctx, anti_affinity, movement_penalty, dry_run): if status == cp_model.OPTIMAL or status == cp_model.FEASIBLE: assignments = {} moves = [] - + for vm_index in range(vms_count): for node_index in range(nodes_count): if solver.Value(vm_to_nodes[(vm_index, node_index)]) == 1: assignments[vm_index] = node_index - + if vm_index in initial_locations and initial_locations[vm_index] != node_index: moves.append((vm_index, initial_locations[vm_index], node_index)) # group_distributions = {} # group_variances = {} # for gid, vm_indices in vm_groups.items(): # if not vm_indices: continue - + # group_dist = [0] * nodes_count # for it in vm_indices: # group_dist[assignments[it]] += 1 - + # group_distributions[gid] = group_dist # group_variances[gid] = np.var(group_dist) - - print(moves) + + print(moves) else: - print('failed') + print("failed") # print(vm_params) # print(initial_locations) # # metrics = proxmox.metrics From 2af2242bfdbb3877a57a95efbf886f0a8f7530e7 Mon Sep 17 00:00:00 2001 From: Yoann Lamouroux Date: Tue, 13 May 2025 15:56:22 +0200 Subject: [PATCH 3/7] display vm migrate commands --- src/pvecontrol/actions/cluster_balance.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/src/pvecontrol/actions/cluster_balance.py b/src/pvecontrol/actions/cluster_balance.py index 7c58b429..03540b60 100644 --- a/src/pvecontrol/actions/cluster_balance.py +++ b/src/pvecontrol/actions/cluster_balance.py @@ -1,3 +1,4 @@ +import sys from itertools import chain import click @@ -21,7 +22,8 @@ def balance(ctx, anti_affinity, movement_penalty, dry_run): overcommit_factor = 1.2 proxmox: PVECluster = PVECluster.create_from_config(ctx.obj["args"].cluster) nodes: list[PVENode] = proxmox.nodes - vms: list[PVEVm] = list(chain.from_iterable(node.vms for node in nodes)) + # keep only non-template instances + vms: list[PVEVm] = list(filter(lambda vm: vm.template == 0, chain.from_iterable(node.vms for node in nodes))) vms_count = len(vms) nodes_count = len(nodes) vm_params = [{"cpu": vm.cpus, "memory": vm.maxmem} for vm in vms] @@ -152,7 +154,9 @@ def balance(ctx, anti_affinity, movement_penalty, dry_run): # group_distributions[gid] = group_dist # group_variances[gid] = np.var(group_dist) - print(moves) + for move in moves: + print(f"{sys.argv[0]} vm migrate --target {nodes[move[2]].node} {vms[move[0]].vmid}") + else: print("failed") # print(vm_params) From 52b6c8e2709e782b7e2d39a17b4406bbfe806028 Mon Sep 17 00:00:00 2001 From: Yoann Lamouroux Date: Tue, 13 May 2025 19:03:40 +0200 Subject: [PATCH 4/7] richer display --- src/pvecontrol/actions/cluster_balance.py | 27 +++++++++++++++++++++-- 1 file changed, 25 insertions(+), 2 deletions(-) diff --git a/src/pvecontrol/actions/cluster_balance.py b/src/pvecontrol/actions/cluster_balance.py index 03540b60..e9d258b0 100644 --- a/src/pvecontrol/actions/cluster_balance.py +++ b/src/pvecontrol/actions/cluster_balance.py @@ -9,6 +9,26 @@ # import numpy as np +def displayer(vms: list[PVEVm], px: PVECluster, moves): + from rich.tree import Tree + from rich import print + from rich.panel import Panel + from rich.columns import Columns + def draw_tree(color): + t = Tree(px.name) + for node in sorted(px.nodes, key=lambda n: n.node): + n = t.add(f"[cyan][b]{node.node}[/b]") + for vm in filter(lambda vm: vm.node == node.node, vms): + if vms.index(vm) in [m[0] for m in moves]: + n.add(f"[{color}]{ vm.name }") + else: + n.add(f"{vm.name}") + return t + before = draw_tree('red') + for move in moves: + vms[move[0]].node = px.nodes[move[2]].node + after = draw_tree('green') + print(Columns([Panel(before), Panel(after)])) @click.command() @click.option( @@ -16,8 +36,9 @@ ) @click.option("-p", "--movement-penalty", default=0, type=int, help="Higher values minimize VM movement") @click.option("-d", "--dry-run", is_flag=True, help="Only print the `vm migrate` commands, doesn't take action") +@click.option("-r", "--rich-display", is_flag=True, help="Use rich display") @click.pass_context -def balance(ctx, anti_affinity, movement_penalty, dry_run): +def balance(ctx, anti_affinity, movement_penalty, dry_run, rich_display): """Balances VMs across nodes in the cluster""" overcommit_factor = 1.2 proxmox: PVECluster = PVECluster.create_from_config(ctx.obj["args"].cluster) @@ -35,7 +56,6 @@ def balance(ctx, anti_affinity, movement_penalty, dry_run): node_map = {vm.name: node_idx for node_idx, node in enumerate(nodes) for vm in node.vms} initial_locations = {i: node_map.get(vm.name) for i, vm in enumerate(vms) if vm.name in node_map} - model = cp_model.CpModel() vm_to_nodes = { (vm_index, node_index): model.NewBoolVar(f"vm_{vm_index}_to_node_{node_index}") @@ -154,6 +174,9 @@ def balance(ctx, anti_affinity, movement_penalty, dry_run): # group_distributions[gid] = group_dist # group_variances[gid] = np.var(group_dist) + if rich_display: + displayer(vms, proxmox, moves) + for move in moves: print(f"{sys.argv[0]} vm migrate --target {nodes[move[2]].node} {vms[move[0]].vmid}") From 48672f82089d43bd9af0f3bc82cfcb3537282821 Mon Sep 17 00:00:00 2001 From: Yoann Lamouroux Date: Wed, 11 Jun 2025 17:41:04 +0200 Subject: [PATCH 5/7] refactor, add timeout, overcommit, better texts, black and lint --- src/pvecontrol/actions/cluster_balance.py | 172 +++++++++++----------- 1 file changed, 87 insertions(+), 85 deletions(-) diff --git a/src/pvecontrol/actions/cluster_balance.py b/src/pvecontrol/actions/cluster_balance.py index e9d258b0..598e11b5 100644 --- a/src/pvecontrol/actions/cluster_balance.py +++ b/src/pvecontrol/actions/cluster_balance.py @@ -1,88 +1,112 @@ import sys from itertools import chain import click +from ortools.sat.python import cp_model + +# displayer requires +import rich +from rich.tree import Tree +from rich.panel import Panel +from rich.columns import Columns from pvecontrol.models.cluster import PVECluster from pvecontrol.models.node import PVENode -from pvecontrol.models.vm import PVEVm -from ortools.sat.python import cp_model +from pvecontrol.models.vm import PVEVm, VmStatus + -# import numpy as np def displayer(vms: list[PVEVm], px: PVECluster, moves): - from rich.tree import Tree - from rich import print - from rich.panel import Panel - from rich.columns import Columns def draw_tree(color): t = Tree(px.name) for node in sorted(px.nodes, key=lambda n: n.node): n = t.add(f"[cyan][b]{node.node}[/b]") - for vm in filter(lambda vm: vm.node == node.node, vms): + node_vms = [vm for vm in vms if vm.node == node.node] + for vm in node_vms: if vms.index(vm) in [m[0] for m in moves]: n.add(f"[{color}]{ vm.name }") else: n.add(f"{vm.name}") return t - before = draw_tree('red') + + before = draw_tree("red") for move in moves: vms[move[0]].node = px.nodes[move[2]].node - after = draw_tree('green') - print(Columns([Panel(before), Panel(after)])) + after = draw_tree("green") + rich.print(Columns([Panel(before), Panel(after)])) + @click.command() +@click.option("-a", "--anti-affinity", default="", help="Balance machines with comma-separated patterns across nodes") +@click.option("-t", "--anti-affinity-tags", default="", help="Balance machines with comma-separated tags across nodes") +@click.option("-p", "--movement-penalty", default=0, type=int, help="Higher values minimize VM movements") +# FIXME: this is the only behavior for now +# @click.option("-d", "--dry-run", is_flag=True, help="Only print the `vm migrate` commands, doesn't take action") +@click.option("-r", "--running", is_flag=True, help="Balance only the currently running instances") +@click.option("-R", "--rich-display", is_flag=True, help="Use rich display") @click.option( - "-a", "--anti-affinity", default="", help="Balance machine with those comma-separated prefixes across nodes" + "--timeout", + type=float, + default=60.0, + help="Limit calculation time, tiny values will result in sub-optimal solution", ) -@click.option("-p", "--movement-penalty", default=0, type=int, help="Higher values minimize VM movement") -@click.option("-d", "--dry-run", is_flag=True, help="Only print the `vm migrate` commands, doesn't take action") -@click.option("-r", "--rich-display", is_flag=True, help="Use rich display") +@click.option("--overcommit", type=float, default=1.0, help="Sets an overcommit ratio regarding CPU nodes requirements") @click.pass_context -def balance(ctx, anti_affinity, movement_penalty, dry_run, rich_display): - """Balances VMs across nodes in the cluster""" - overcommit_factor = 1.2 +# def balance(ctx, anti_affinity, anti_affinity_tags, movement_penalty, dry_run, running, rich_display): +def balance(ctx, anti_affinity, anti_affinity_tags, movement_penalty, running, rich_display, timeout, overcommit): + """Balances VMs across nodes in the cluster, bear in mind the model used won't give the same result upon every calls""" + proxmox: PVECluster = PVECluster.create_from_config(ctx.obj["args"].cluster) nodes: list[PVENode] = proxmox.nodes # keep only non-template instances vms: list[PVEVm] = list(filter(lambda vm: vm.template == 0, chain.from_iterable(node.vms for node in nodes))) - vms_count = len(vms) - nodes_count = len(nodes) + + if running: + vms = list(filter(lambda vm: vm.status == VmStatus.RUNNING, vms)) + vm_params = [{"cpu": vm.cpus, "memory": vm.maxmem} for vm in vms] - node_caps = list({"cpu": node.maxcpu, "memory": node.maxmem} for node in nodes) - vm_groups = { - aag: list(map(lambda i: i[0], filter(lambda x: aag in x[1].name, enumerate(vms)))) - for aag in anti_affinity.split(",") - } + node_caps = [{"cpu": node.maxcpu, "memory": node.maxmem} for node in nodes] + + # we build anti-affinity as such : {'group_name': [list of vm indices with matching name]} + vm_antiaffinity_groups = {} + for aag in anti_affinity.split(","): + vm_antiaffinity_groups[aag] = [i for i, vm in enumerate(vms) if aag in vm.name] + for aat in anti_affinity_tags.split(","): + vm_antiaffinity_groups[aat] = [i for i, vm in enumerate(vms) if aat in vm.tags] + + node_map = {vm.vmid: node_idx for node_idx, node in enumerate(nodes) for vm in node.vms} + initial_locations = {i: node_map.get(vm.vmid) for i, vm in enumerate(vms) if vm.vmid in node_map} - node_map = {vm.name: node_idx for node_idx, node in enumerate(nodes) for vm in node.vms} - initial_locations = {i: node_map.get(vm.name) for i, vm in enumerate(vms) if vm.name in node_map} model = cp_model.CpModel() + + # list all possible placement in this cluster (vm * node) vm_to_nodes = { (vm_index, node_index): model.NewBoolVar(f"vm_{vm_index}_to_node_{node_index}") - for vm_index in range(vms_count) - for node_index in range(nodes_count) + for vm_index, _ in enumerate(vms) + for node_index, _ in enumerate(nodes) } - for vm_index in range(vms_count): - model.Add(sum(vm_to_nodes[(vm_index, node_index)] for node_index in range(nodes_count)) == 1) + # a given vm should exist in *only one* node + for vm_index, _ in enumerate(vms): + model.Add(sum(vm_to_nodes[(vm_index, node_index)] for node_index, _ in enumerate(nodes)) == 1) - for node_index in range(nodes_count): - overcomitted_cpu_cap = int(node_caps[node_index]["cpu"] * overcommit_factor) + # a node can allocate only its overcommit*cpus, and only its ram + for node_index, _ in enumerate(nodes): + overcomitted_cpu_cap = int(node_caps[node_index]["cpu"] * overcommit) resources = {"cpu": overcomitted_cpu_cap, "memory": node_caps[node_index]["memory"]} for k, v in resources.items(): - model.Add( - sum(vm_params[vm_index][k] * vm_to_nodes[(vm_index, node_index)] for vm_index in range(vms_count)) <= v - ) + model.Add(sum(vm_params[i][k] * vm_to_nodes[(i, node_index)] for i, _ in enumerate(vms)) <= v) obj_terms = [] - for gid, vm_indices in vm_groups.items(): + # there should be as few antiaffinity groups members per node as possible (this implies they'll be spread as + # much as possible) + for gid, vm_indices in vm_antiaffinity_groups.items(): if not vm_indices: continue max_vms_in_node = model.NewIntVar(0, len(vm_indices), f"max_vms_from_group_{gid}") group_vms_per_node = {} - for node_index in range(nodes_count): + for node_index, _ in enumerate(nodes): group_vms_per_node[node_index] = model.NewIntVar( 0, len(vm_indices), f"group_{gid}_vms_in_node_{node_index}" ) @@ -92,8 +116,8 @@ def balance(ctx, anti_affinity, movement_penalty, dry_run, rich_display): model.Add(max_vms_in_node >= group_vms_per_node[node_index]) obj_terms.append(max_vms_in_node * len(vm_indices) * 100) - for node_index_start in range(nodes_count): - for node_index_remain in range(node_index_start + 1, nodes_count): + for node_index_start, _ in enumerate(nodes): + for node_index_remain in range(node_index_start + 1, len(nodes)): diff = model.NewIntVar( 0, len(vm_indices), f"diff_group_{gid}_nodes_{node_index_start}_{node_index_remain}" ) @@ -101,7 +125,8 @@ def balance(ctx, anti_affinity, movement_penalty, dry_run, rich_display): obj_terms.append(diff * 10) vm_moved = {} - for vm_index in range(vms_count): + # we add the movement penalty to our contraints, the solution will carry "non-moving" VM account for these + for vm_index, _ in enumerate(vms): if node_index in initial_locations: vm_moved[vm_index] = model.NewBoolVar(f"vm_{vm_index}_moved") initial_node = initial_locations[vm_index] @@ -109,70 +134,51 @@ def balance(ctx, anti_affinity, movement_penalty, dry_run, rich_display): model.Add(vm_to_nodes[(vm_index, initial_node)] == 0).OnlyEnforceIf(vm_moved[vm_index]) obj_terms.append(vm_moved[vm_index] * movement_penalty) - resource_per_node = {} - max_resource = {} + # even when overcommitting, we deal with discrete cpu core counts resources = { "cpu": { - "max_cap": int(max(cap["cpu"] for cap in node_caps) * overcommit_factor), - "node_cap": lambda node_index: int(node_caps[node_index]["cpu"] * overcommit_factor), + "max_cap": int(max(cap["cpu"] for cap in node_caps) * overcommit), + "node_cap": lambda node_index: int(node_caps[node_index]["cpu"] * overcommit), }, "memory": { "max_cap": max(cap["memory"] for cap in node_caps), "node_cap": lambda node_index: node_caps[node_index]["memory"], }, } + + resource_per_node = {} + max_resource = {} for rtype, rvalue in resources.items(): resource_per_node[rtype] = {} - for node_index in range(nodes_count): + for node_index, _ in enumerate(nodes): node_var = model.NewIntVar(0, rvalue["node_cap"](node_index), f"{rtype}_in_node_{node_index}") resource_per_node[rtype][node_index] = node_var - model.Add( - node_var - == sum( - vm_params[vm_index][rtype] * vm_to_nodes[(vm_index, node_index)] for vm_index in range(vms_count) - ) - ) + model.Add(node_var == sum(vm_params[i][rtype] * vm_to_nodes[(i, node_index)] for i, _ in enumerate(vms))) max_resource[rtype] = model.NewIntVar(0, rvalue["max_cap"], f"max_{rtype}") - max_cpu = max_resource["cpu"] - max_memory = max_resource["memory"] - cpu_per_node = resource_per_node["cpu"] - memory_per_node = resource_per_node["memory"] - - for node_index in range(nodes_count): - model.Add(max_cpu >= cpu_per_node[node_index]) - model.Add(max_memory >= memory_per_node[node_index]) + for node_index, _ in enumerate(nodes): + model.Add(max_resource["cpu"] >= resource_per_node["cpu"][node_index]) + model.Add(max_resource["memory"] >= resource_per_node["memory"][node_index]) - obj_terms.append(max_cpu) - obj_terms.append(max_memory) + obj_terms.append(max_resource["cpu"]) + obj_terms.append(max_resource["memory"]) model.Minimize(sum(obj_terms)) solver = cp_model.CpSolver() - solver.parameters.max_time_in_seconds = 60.0 - status = solver.Solve(model=model) - if status == cp_model.OPTIMAL or status == cp_model.FEASIBLE: + + solver.parameters.max_time_in_seconds = timeout + if solver.Solve(model=model) in (cp_model.OPTIMAL, cp_model.FEASIBLE): assignments = {} moves = [] - for vm_index in range(vms_count): - for node_index in range(nodes_count): + for vm_index, _ in enumerate(vms): + for node_index, _ in enumerate(nodes): if solver.Value(vm_to_nodes[(vm_index, node_index)]) == 1: assignments[vm_index] = node_index if vm_index in initial_locations and initial_locations[vm_index] != node_index: moves.append((vm_index, initial_locations[vm_index], node_index)) - # group_distributions = {} - # group_variances = {} - # for gid, vm_indices in vm_groups.items(): - # if not vm_indices: continue - - # group_dist = [0] * nodes_count - # for it in vm_indices: - # group_dist[assignments[it]] += 1 - - # group_distributions[gid] = group_dist - # group_variances[gid] = np.var(group_dist) if rich_display: displayer(vms, proxmox, moves) @@ -181,10 +187,6 @@ def balance(ctx, anti_affinity, movement_penalty, dry_run, rich_display): print(f"{sys.argv[0]} vm migrate --target {nodes[move[2]].node} {vms[move[0]].vmid}") else: - print("failed") - # print(vm_params) - # print(initial_locations) - # # metrics = proxmox.metrics - # print(anti_affinity) - # print(movement_penalty) - # print(dry_run) + print( + "# an optimal couldn't be found, give it more time (using -t), set a bigger overcommit, provision more nodes" + ) From 325a28ca7313cf621534674dca0fe870a7b782a8 Mon Sep 17 00:00:00 2001 From: Yoann Lamouroux Date: Thu, 12 Jun 2025 14:02:28 +0200 Subject: [PATCH 6/7] fix pylint (lot of factoring) --- src/pvecontrol/actions/cluster_balance.py | 201 ++++++++++++++-------- 1 file changed, 128 insertions(+), 73 deletions(-) diff --git a/src/pvecontrol/actions/cluster_balance.py b/src/pvecontrol/actions/cluster_balance.py index 598e11b5..d589b36b 100644 --- a/src/pvecontrol/actions/cluster_balance.py +++ b/src/pvecontrol/actions/cluster_balance.py @@ -10,11 +10,9 @@ from rich.columns import Columns from pvecontrol.models.cluster import PVECluster -from pvecontrol.models.node import PVENode from pvecontrol.models.vm import PVEVm, VmStatus - def displayer(vms: list[PVEVm], px: PVECluster, moves): def draw_tree(color): t = Tree(px.name) @@ -35,51 +33,9 @@ def draw_tree(color): rich.print(Columns([Panel(before), Panel(after)])) -@click.command() -@click.option("-a", "--anti-affinity", default="", help="Balance machines with comma-separated patterns across nodes") -@click.option("-t", "--anti-affinity-tags", default="", help="Balance machines with comma-separated tags across nodes") -@click.option("-p", "--movement-penalty", default=0, type=int, help="Higher values minimize VM movements") -# FIXME: this is the only behavior for now -# @click.option("-d", "--dry-run", is_flag=True, help="Only print the `vm migrate` commands, doesn't take action") -@click.option("-r", "--running", is_flag=True, help="Balance only the currently running instances") -@click.option("-R", "--rich-display", is_flag=True, help="Use rich display") -@click.option( - "--timeout", - type=float, - default=60.0, - help="Limit calculation time, tiny values will result in sub-optimal solution", -) -@click.option("--overcommit", type=float, default=1.0, help="Sets an overcommit ratio regarding CPU nodes requirements") -@click.pass_context -# def balance(ctx, anti_affinity, anti_affinity_tags, movement_penalty, dry_run, running, rich_display): -def balance(ctx, anti_affinity, anti_affinity_tags, movement_penalty, running, rich_display, timeout, overcommit): - """Balances VMs across nodes in the cluster, bear in mind the model used won't give the same result upon every calls""" - - proxmox: PVECluster = PVECluster.create_from_config(ctx.obj["args"].cluster) - nodes: list[PVENode] = proxmox.nodes - # keep only non-template instances - vms: list[PVEVm] = list(filter(lambda vm: vm.template == 0, chain.from_iterable(node.vms for node in nodes))) - - if running: - vms = list(filter(lambda vm: vm.status == VmStatus.RUNNING, vms)) - - vm_params = [{"cpu": vm.cpus, "memory": vm.maxmem} for vm in vms] - node_caps = [{"cpu": node.maxcpu, "memory": node.maxmem} for node in nodes] - - # we build anti-affinity as such : {'group_name': [list of vm indices with matching name]} - vm_antiaffinity_groups = {} - for aag in anti_affinity.split(","): - vm_antiaffinity_groups[aag] = [i for i, vm in enumerate(vms) if aag in vm.name] - for aat in anti_affinity_tags.split(","): - vm_antiaffinity_groups[aat] = [i for i, vm in enumerate(vms) if aat in vm.tags] - - node_map = {vm.vmid: node_idx for node_idx, node in enumerate(nodes) for vm in node.vms} - initial_locations = {i: node_map.get(vm.vmid) for i, vm in enumerate(vms) if vm.vmid in node_map} - - model = cp_model.CpModel() - +def __vm_to_nodes(model, nodes, vms, node_caps, vm_params, overcommit): # list all possible placement in this cluster (vm * node) - vm_to_nodes = { + vm_to_nodes: dict[tuple[int, int], cp_model.IntVar] = { (vm_index, node_index): model.NewBoolVar(f"vm_{vm_index}_to_node_{node_index}") for vm_index, _ in enumerate(vms) for node_index, _ in enumerate(nodes) @@ -95,8 +51,11 @@ def balance(ctx, anti_affinity, anti_affinity_tags, movement_penalty, running, r resources = {"cpu": overcomitted_cpu_cap, "memory": node_caps[node_index]["memory"]} for k, v in resources.items(): model.Add(sum(vm_params[i][k] * vm_to_nodes[(i, node_index)] for i, _ in enumerate(vms)) <= v) + return vm_to_nodes - obj_terms = [] + +def __constraints_antiaffinities_nodes(model, nodes, vm_to_nodes, vm_antiaffinity_groups): + obj_terms: list[cp_model.IntVar] = [] # there should be as few antiaffinity groups members per node as possible (this implies they'll be spread as # much as possible) for gid, vm_indices in vm_antiaffinity_groups.items(): @@ -123,18 +82,13 @@ def balance(ctx, anti_affinity, anti_affinity_tags, movement_penalty, running, r ) model.AddAbsEquality(diff, group_vms_per_node[node_index_start] - group_vms_per_node[node_index_remain]) obj_terms.append(diff * 10) + return obj_terms - vm_moved = {} - # we add the movement penalty to our contraints, the solution will carry "non-moving" VM account for these - for vm_index, _ in enumerate(vms): - if node_index in initial_locations: - vm_moved[vm_index] = model.NewBoolVar(f"vm_{vm_index}_moved") - initial_node = initial_locations[vm_index] - model.Add(vm_to_nodes[(vm_index, initial_node)] == 1).OnlyEnforceIf(vm_moved[vm_index].Not()) - model.Add(vm_to_nodes[(vm_index, initial_node)] == 0).OnlyEnforceIf(vm_moved[vm_index]) - obj_terms.append(vm_moved[vm_index] * movement_penalty) - # even when overcommitting, we deal with discrete cpu core counts +def __constraints_resources_vm( + model, nodes, vms, initial_locations, vm_to_nodes, overcommit, node_caps, vm_params, movement_penalty +): + obj_terms = [] resources = { "cpu": { "max_cap": int(max(cap["cpu"] for cap in node_caps) * overcommit), @@ -146,28 +100,47 @@ def balance(ctx, anti_affinity, anti_affinity_tags, movement_penalty, running, r }, } - resource_per_node = {} - max_resource = {} - for rtype, rvalue in resources.items(): - resource_per_node[rtype] = {} + def __gather_movements(): + vm_moved = {} + # we add the movement penalty to our contraints, the solution will carry "non-moving" VM account for these for node_index, _ in enumerate(nodes): - node_var = model.NewIntVar(0, rvalue["node_cap"](node_index), f"{rtype}_in_node_{node_index}") - resource_per_node[rtype][node_index] = node_var - model.Add(node_var == sum(vm_params[i][rtype] * vm_to_nodes[(i, node_index)] for i, _ in enumerate(vms))) + for vm_index, _ in enumerate(vms): + if node_index in initial_locations: + vm_moved[vm_index] = model.NewBoolVar(f"vm_{vm_index}_moved") + initial_node = initial_locations[vm_index] + model.Add(vm_to_nodes[(vm_index, initial_node)] == 1).OnlyEnforceIf(vm_moved[vm_index].Not()) + model.Add(vm_to_nodes[(vm_index, initial_node)] == 0).OnlyEnforceIf(vm_moved[vm_index]) + obj_terms.append(vm_moved[vm_index] * movement_penalty) - max_resource[rtype] = model.NewIntVar(0, rvalue["max_cap"], f"max_{rtype}") + # even when overcommitting, we deal with discrete cpu core counts + def __constraint_resources(): + resource_per_node = {} + max_resource = {} + for rtype, rvalue in resources.items(): + resource_per_node[rtype] = {} + for node_index, _ in enumerate(nodes): + node_var = model.NewIntVar(0, rvalue["node_cap"](node_index), f"{rtype}_in_node_{node_index}") + resource_per_node[rtype][node_index] = node_var + model.Add( + node_var == sum(vm_params[i][rtype] * vm_to_nodes[(i, node_index)] for i, _ in enumerate(vms)) + ) - for node_index, _ in enumerate(nodes): - model.Add(max_resource["cpu"] >= resource_per_node["cpu"][node_index]) - model.Add(max_resource["memory"] >= resource_per_node["memory"][node_index]) + max_resource[rtype] = model.NewIntVar(0, rvalue["max_cap"], f"max_{rtype}") - obj_terms.append(max_resource["cpu"]) - obj_terms.append(max_resource["memory"]) + for node_index, _ in enumerate(nodes): + model.Add(max_resource["cpu"] >= resource_per_node["cpu"][node_index]) + model.Add(max_resource["memory"] >= resource_per_node["memory"][node_index]) - model.Minimize(sum(obj_terms)) - solver = cp_model.CpSolver() + obj_terms.append(max_resource["cpu"]) + obj_terms.append(max_resource["memory"]) - solver.parameters.max_time_in_seconds = timeout + __gather_movements() + __constraint_resources() + + return obj_terms + + +def __solve_model(solver, model, vms, nodes, initial_locations, vm_to_nodes, rich_display, proxmox): if solver.Solve(model=model) in (cp_model.OPTIMAL, cp_model.FEASIBLE): assignments = {} moves = [] @@ -190,3 +163,85 @@ def balance(ctx, anti_affinity, anti_affinity_tags, movement_penalty, running, r print( "# an optimal couldn't be found, give it more time (using -t), set a bigger overcommit, provision more nodes" ) + + +# we build anti-affinity as such : {'group_name': [list of vm indices with matching name]} +def __gather_antiaffinity_groups(vms, anti_affinity, anti_affinity_tags): + vm_antiaffinity_groups = {} + for aag in anti_affinity.split(","): + vm_antiaffinity_groups[aag] = [i for i, vm in enumerate(vms) if aag in vm.name] + for aat in anti_affinity_tags.split(","): + vm_antiaffinity_groups[aat] = [i for i, vm in enumerate(vms) if aat in vm.tags] + return vm_antiaffinity_groups + + +def __nodes_map(nodes): + return {vm.vmid: node_idx for node_idx, node in enumerate(nodes) for vm in node.vms} + + +@click.command() +@click.option("-a", "--anti-affinity", default="", help="Balance machines with comma-separated patterns across nodes") +@click.option("-t", "--anti-affinity-tags", default="", help="Balance machines with comma-separated tags across nodes") +@click.option("-p", "--movement-penalty", default=0, type=int, help="Higher values minimize VM movements") +# FIXME: this is the only behavior for now +# @click.option("-d", "--dry-run", is_flag=True, help="Only print the `vm migrate` commands, doesn't take action") +@click.option("-r", "--running", is_flag=True, help="Balance only the currently running instances") +@click.option("-R", "--rich-display", is_flag=True, help="Use rich display") +@click.option( + "--timeout", + type=float, + default=60.0, + help="Limit calculation time, tiny values will result in sub-optimal solution", +) +@click.option("--overcommit", type=float, default=1.0, help="Sets an overcommit ratio regarding CPU nodes requirements") +@click.pass_context +# def balance(ctx, anti_affinity, anti_affinity_tags, movement_penalty, dry_run, running, rich_display): +def balance(ctx, anti_affinity, anti_affinity_tags, movement_penalty, running, rich_display, timeout, overcommit): + """Balances VMs across nodes in the cluster, bear in mind the model used won't give the same result upon every calls""" + proxmox: PVECluster = PVECluster.create_from_config(ctx.obj["args"].cluster) + # keep only non-template instances + vms: list[PVEVm] = list( + filter(lambda vm: vm.template == 0, chain.from_iterable(node.vms for node in proxmox.nodes)) + ) + if running: + vms = list(filter(lambda vm: vm.status == VmStatus.RUNNING, vms)) + + vm_antiaffinity_groups = __gather_antiaffinity_groups(vms, anti_affinity, anti_affinity_tags) + nodes_map = __nodes_map(proxmox.nodes) + model = cp_model.CpModel() + vm_to_nodes = __vm_to_nodes( + model, + proxmox.nodes, + vms, + [{"cpu": node.maxcpu, "memory": node.maxmem} for node in proxmox.nodes], + [{"cpu": vm.cpus, "memory": vm.maxmem} for vm in vms], + overcommit, + ) + model.Minimize( + sum( + __constraints_antiaffinities_nodes(model, proxmox.nodes, vm_to_nodes, vm_antiaffinity_groups) + + __constraints_resources_vm( + model, + proxmox.nodes, + vms, + {i: nodes_map.get(vm.vmid) for i, vm in enumerate(vms) if vm.vmid in nodes_map}, + vm_to_nodes, + overcommit, + [{"cpu": node.maxcpu, "memory": node.maxmem} for node in proxmox.nodes], + [{"cpu": vm.cpus, "memory": vm.maxmem} for vm in vms], + movement_penalty, + ) + ) + ) + solver = cp_model.CpSolver() + solver.parameters.max_time_in_seconds = timeout + __solve_model( + solver, + model, + vms, + proxmox.nodes, + {i: nodes_map.get(vm.vmid) for i, vm in enumerate(vms) if vm.vmid in nodes_map}, + vm_to_nodes, + rich_display, + proxmox, + ) From 12b3f76d5d381a2d5b397c443fc941e63ebfd604 Mon Sep 17 00:00:00 2001 From: Yoann Lamouroux Date: Thu, 12 Jun 2025 14:11:34 +0200 Subject: [PATCH 7/7] missing rich dep --- requirements.txt | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/requirements.txt b/requirements.txt index 2903658f..0d664ddf 100644 --- a/requirements.txt +++ b/requirements.txt @@ -12,4 +12,5 @@ requests requests_toolbelt openssh_wrapper paramiko -ortools \ No newline at end of file +ortools +rich \ No newline at end of file