From 59c8ea3f8fcca8e75bd2e383db256b0024bc6744 Mon Sep 17 00:00:00 2001 From: chrisfu Date: Mon, 23 Mar 2026 00:06:47 -0700 Subject: [PATCH] Add CNPG placement planner with persistence and integration tests - Introduce `cnpg_placement.py` to handle round-robin node assignments for CloudNativePG clusters. - Implement persistent placement plans with schema normalization and hashing. - Add `load_cnpg_placement_plan` and `save_cnpg_placement_plan` for plan persistence. - Integrate planner with installer to support node-based topology configuration. - Update `actions.py` with placement planning logic, including rebalance support and node eligibility checks. - Extend shell utilities (`init_cloudnative_pg.sh`) for placement-aware CNPG topology adjustments. - Add comprehensive unit tests and integration tests for planner functionality, persistence, and shell environment exports. --- etc/init_cloudnative_pg.sh | 124 +++++++++++++++- knoe/core/actions.py | 158 ++++++++++++++++++++ knoe/core/cnpg_placement.py | 186 ++++++++++++++++++++++++ tests/installer/test_actions_helpers.py | 19 +++ tests/installer/test_cnpg_placement.py | 107 ++++++++++++++ 5 files changed, 593 insertions(+), 1 deletion(-) create mode 100644 knoe/core/cnpg_placement.py create mode 100644 tests/installer/test_cnpg_placement.py diff --git a/etc/init_cloudnative_pg.sh b/etc/init_cloudnative_pg.sh index 01c6c79..5ebc077 100755 --- a/etc/init_cloudnative_pg.sh +++ b/etc/init_cloudnative_pg.sh @@ -78,6 +78,12 @@ BACKUP_WAIT_TIMEOUT=${BACKUP_WAIT_TIMEOUT:-1800} CNPG_TOPOLOGY=${CNPG_TOPOLOGY:-auto} CNPG_STAGE1_NODE=${CNPG_STAGE1_NODE:-myrddin.prole.org} CNPG_DB_NODE_SELECTOR=${CNPG_DB_NODE_SELECTOR:-} +CNPG_PLACEMENT_PLAN_FILE=${CNPG_PLACEMENT_PLAN_FILE:-} +CNPG_PLACEMENT_ELIGIBLE_NODES=${CNPG_PLACEMENT_ELIGIBLE_NODES:-} +CNPG_PLACEMENT_PLAN_ID=${CNPG_PLACEMENT_PLAN_ID:-} +CNPG_PLACEMENT_PLAN_HASH=${CNPG_PLACEMENT_PLAN_HASH:-} +CNPG_PLACEMENT_REUSED=${CNPG_PLACEMENT_REUSED:-} +CNPG_PLACEMENT_ASSIGNMENTS=${CNPG_PLACEMENT_ASSIGNMENTS:-} CNPG_ALLOW_COLOCATED_INSTANCES=${CNPG_ALLOW_COLOCATED_INSTANCES:-} CNPG_POD_ANTI_AFFINITY_TYPE=${CNPG_POD_ANTI_AFFINITY_TYPE:-} @@ -1887,7 +1893,77 @@ _cnpg_node_exists() { [[ "$out" == "$node_name" ]] } +_cnpg_planner_nodes_csv() { + if [[ -n "${CNPG_PLACEMENT_ELIGIBLE_NODES:-}" ]]; then + echo "$CNPG_PLACEMENT_ELIGIBLE_NODES" + return 0 + fi + if [[ -n "${CNPG_PLACEMENT_PLAN_FILE:-}" && -f "${CNPG_PLACEMENT_PLAN_FILE}" ]]; then + jq -r '.eligible_nodes // [] | map(tostring) | join(",")' "$CNPG_PLACEMENT_PLAN_FILE" 2>/dev/null || true + return 0 + fi + echo "" +} + +_cnpg_planner_nodes_lines() { + local raw + raw=$(_cnpg_planner_nodes_csv) + printf '%s' "$raw" \ + | tr ',' '\n' \ + | sed 's/^[[:space:]]*//; s/[[:space:]]*$//' \ + | awk 'NF && !seen[$0]++ { print $0 }' +} + +_cnpg_has_planner_nodes() { + [[ -n "$(_cnpg_planner_nodes_lines | head -n 1)" ]] +} + +_cnpg_planner_first_node() { + _cnpg_planner_nodes_lines | head -n 1 +} + +_cnpg_planner_nodes_count() { + _cnpg_planner_nodes_lines | wc -l | tr -d ' ' +} + +_cnpg_planner_nodes_json() { + local first=1 + local out='[' + local node esc + while IFS= read -r node || [[ -n "$node" ]]; do + [[ -n "$node" ]] || continue + esc=${node//\\/\\\\} + esc=${esc//\"/\\\"} + if [[ "$first" -eq 1 ]]; then + first=0 + else + out+=',' + fi + out+="\"${esc}\"" + done < <(_cnpg_planner_nodes_lines) + out+=']' + echo "$out" +} + +_ready_schedulable_planner_nodes_count() { + local count=0 node ready unsched + while IFS= read -r node || [[ -n "$node" ]]; do + [[ -n "$node" ]] || continue + ready=$(kubectl get node "$node" -o jsonpath='{range .status.conditions[?(@.type=="Ready")]}{.status}{end}' 2>/dev/null || true) + unsched=$(kubectl get node "$node" -o jsonpath='{.spec.unschedulable}' 2>/dev/null || true) + if [[ "$ready" == "True" && "$unsched" != "true" ]]; then + count=$((count + 1)) + fi + done < <(_cnpg_planner_nodes_lines) + echo "$count" +} + _cnpg_effective_topology() { + if _cnpg_has_planner_nodes; then + echo "planner" + return 0 + fi + local t="${CNPG_TOPOLOGY:-auto}" case "$t" in stage1|stage-1|stage1-single-node|single-node) @@ -1914,6 +1990,16 @@ _cnpg_effective_topology() { } _cnpg_effective_node_selector() { + if _cnpg_has_planner_nodes; then + local planner_count planner_first + planner_count=$(_coerce_uint "$(_cnpg_planner_nodes_count)" 0) + planner_first=$(_cnpg_planner_first_node) + if (( planner_count == 1 )) && [[ -n "$planner_first" ]]; then + echo "kubernetes.io/hostname=${planner_first}" + return 0 + fi + fi + if [[ -n "${CNPG_DB_NODE_SELECTOR:-}" ]]; then echo "$CNPG_DB_NODE_SELECTOR" return 0 @@ -1946,8 +2032,22 @@ patch_cnpg_manifest_for_topology() { # relax anti-affinity so all instances can schedule on one node. local manifest_file="$1" [[ -f "$manifest_file" ]] || return 0 + local planner_single_node=0 - if [[ "$(_cnpg_effective_topology)" != "stage1" ]]; then + if _cnpg_has_planner_nodes; then + local planner_count planner_node + planner_count=$(_coerce_uint "$(_cnpg_planner_nodes_count)" 0) + if (( planner_count != 1 )); then + return 0 + fi + planner_node=$(_cnpg_planner_first_node) + if [[ -n "$planner_node" ]]; then + CNPG_STAGE1_NODE="$planner_node" + planner_single_node=1 + fi + fi + + if (( planner_single_node == 0 )) && [[ "$(_cnpg_effective_topology)" != "stage1" ]]; then return 0 fi @@ -1986,6 +2086,23 @@ ensure_cnpg_cluster_scheduling() { fi fi + if _cnpg_has_planner_nodes; then + local planner_count nodes_json + planner_count=$(_coerce_uint "$(_cnpg_planner_nodes_count)" 0) + if (( planner_count > 0 )); then + if [[ -z "${CNPG_POD_ANTI_AFFINITY_TYPE:-}" ]]; then + if (( planner_count <= 1 )) || _cnpg_allow_colocated_instances; then + anti="preferred" + else + anti="required" + fi + fi + nodes_json=$(_cnpg_planner_nodes_json) + kubectl -n "$NAMESPACE" patch cluster "$CNPG_CLUSTER_NAME" --type merge -p "{\"spec\":{\"affinity\":{\"enablePodAntiAffinity\":true,\"podAntiAffinityType\":\"${anti}\",\"topologyKey\":\"kubernetes.io/hostname\",\"nodeAffinity\":{\"requiredDuringSchedulingIgnoredDuringExecution\":{\"nodeSelectorTerms\":[{\"matchExpressions\":[{\"key\":\"kubernetes.io/hostname\",\"operator\":\"In\",\"values\":${nodes_json}}]}]}}}}}" >/dev/null 2>&1 || true + return 0 + fi + fi + kubectl -n "$NAMESPACE" patch cluster "$CNPG_CLUSTER_NAME" --type merge -p "{\"spec\":{\"affinity\":{\"enablePodAntiAffinity\":true,\"podAntiAffinityType\":\"${anti}\",\"topologyKey\":\"kubernetes.io/hostname\",\"nodeAffinity\":{\"requiredDuringSchedulingIgnoredDuringExecution\":{\"nodeSelectorTerms\":[{\"matchExpressions\":[{\"key\":\"${key}\",\"operator\":\"In\",\"values\":[\"${val}\"]}]}]}}}}}}" >/dev/null 2>&1 || true } @@ -1995,6 +2112,11 @@ _ready_schedulable_nodes_count() { } _ready_schedulable_db_nodes_count() { + if _cnpg_has_planner_nodes; then + _ready_schedulable_planner_nodes_count + return 0 + fi + local selector selector=$(_cnpg_effective_node_selector) kubectl get nodes -l "$selector" --no-headers 2>/dev/null \ diff --git a/knoe/core/actions.py b/knoe/core/actions.py index 447ca92..d4d4277 100644 --- a/knoe/core/actions.py +++ b/knoe/core/actions.py @@ -82,6 +82,11 @@ from knoe.core.milestones import ( ) from knoe.core.stream_exec import run_streaming_cmd from knoe.core.build_context import copy_build_context_dir +from knoe.core.cnpg_placement import ( + load_cnpg_placement_plan, + plan_cnpg_placement, + save_cnpg_placement_plan, +) from knoe.core.policy import ( POLICY_CFG_KEY, OPTIONAL_WORKLOADS_MIN_READY_SCHEDULABLE_NODES, @@ -1587,8 +1592,150 @@ class ProleInstaller: if reg_internal: env["LOCAL_REGISTRY_INTERNAL"] = reg_internal + try: + placement_plan, placement_path = self._resolve_cnpg_placement_plan(namespace) + eligible_nodes = placement_plan.get("eligible_nodes") or [] + assignments = placement_plan.get("assignments") or {} + assignment_pairs = [] + for ordinal in sorted(assignments.keys(), key=lambda k: int(str(k))): + assignment_pairs.append(f"{ordinal}:{assignments[ordinal]}") + + env["CNPG_PLACEMENT_PLAN_FILE"] = str(placement_path) + env["CNPG_PLACEMENT_PLAN_ID"] = str(placement_plan.get("plan_id") or "") + env["CNPG_PLACEMENT_PLAN_HASH"] = str( + placement_plan.get("plan_hash") or "" + ) + env["CNPG_PLACEMENT_ELIGIBLE_NODES"] = ",".join(eligible_nodes) + env["CNPG_PLACEMENT_ASSIGNMENTS"] = ",".join(assignment_pairs) + env["CNPG_PLACEMENT_REUSED"] = _bool_str( + bool((placement_plan.get("metadata") or {}).get("reused", False)) + ) + env.setdefault( + "CNPG_CLUSTER_NAME", + str(placement_plan.get("cluster_name") or "knoe-db"), + ) + env.setdefault( + "CNPG_INSTANCES", + str(placement_plan.get("desired_instances") or "3"), + ) + + # Compatibility bridge for existing shell topology flow. + if eligible_nodes: + env.setdefault("CNPG_STAGE1_NODE", str(eligible_nodes[0])) + if len(eligible_nodes) == 1: + env.setdefault( + "CNPG_DB_NODE_SELECTOR", + f"kubernetes.io/hostname={eligible_nodes[0]}", + ) + except Exception as e: + self.err(f"[WARN] CNPG placement planning failed; using fallback topology: {e}") + return env + def _cnpg_cluster_name(self) -> str: + glob = self.prole_cfg_data.get("Global", {}) or {} + cluster_name = ( + os.environ.get("CNPG_CLUSTER_NAME") + or str(glob.get("CNPG_CLUSTER_NAME") or "") + ).strip() + return cluster_name or "knoe-db" + + def _cnpg_desired_instances(self) -> int: + glob = self.prole_cfg_data.get("Global", {}) or {} + raw = (os.environ.get("CNPG_INSTANCES") or str(glob.get("CNPG_INSTANCES") or "")).strip() + try: + val = int(raw) if raw else 3 + except Exception: + val = 3 + return max(1, val) + + def _cnpg_rebalance_requested(self) -> bool: + glob = self.prole_cfg_data.get("Global", {}) or {} + raw = ( + os.environ.get("CNPG_REBALANCE") + or str(glob.get("CNPG_REBALANCE") or "") + ) + return _parse_bool(raw, default=False) + + def _cnpg_candidate_nodes(self) -> list[str]: + glob = self.prole_cfg_data.get("Global", {}) or {} + nodes: set[str] = set() + + explicit = ( + os.environ.get("CNPG_ELIGIBLE_NODES") + or str(glob.get("CNPG_ELIGIBLE_NODES") or "") + ) + if explicit: + for part in explicit.split(","): + node = str(part or "").strip() + if node: + nodes.add(node) + + mounts_raw = str( + (self.prole_cfg_data.get("Storage", {}) or {}).get("ANSIBLE_ISCSI_MOUNTS", "") + ).strip() + if mounts_raw: + try: + mounts = json.loads(mounts_raw) + except Exception: + mounts = {} + if isinstance(mounts, dict): + for details in mounts.values(): + if not isinstance(details, dict): + continue + node = str(details.get("host") or "").strip() + if node: + nodes.add(node) + + stage1 = str(glob.get("CNPG_STAGE1_NODE") or "").strip() + if stage1: + nodes.add(stage1) + + return sorted(nodes) + + def _cnpg_plan_path(self, namespace: str, cluster_name: str) -> Path: + conf_root = self._get_input( + "env_setup.PROLE_CONF", + str((getattr(self, "project_root", None) or PROJECT_ROOT) / "conf"), + ) + conf_dir = Path(conf_root) + + def _safe_segment(raw: str) -> str: + s = str(raw or "").strip() + if not s: + return "default" + return "".join(ch if ch.isalnum() or ch in "-_." else "-" for ch in s) + + ns_seg = _safe_segment(namespace) + cluster_seg = _safe_segment(cluster_name) + return conf_dir / "cnpg-placement" / f"{ns_seg}-{cluster_seg}.json" + + def _resolve_cnpg_placement_plan(self, namespace: str) -> tuple[dict, Path]: + cluster_name = self._cnpg_cluster_name() + desired_instances = self._cnpg_desired_instances() + candidate_nodes = self._cnpg_candidate_nodes() + rebalance = self._cnpg_rebalance_requested() + plan_path = self._cnpg_plan_path(namespace, cluster_name) + prior_plan = load_cnpg_placement_plan(plan_path) + + placement_plan = plan_cnpg_placement( + cluster_name=cluster_name, + desired_instances=desired_instances, + candidate_nodes=candidate_nodes, + prior_plan=prior_plan, + rebalance=rebalance, + ) + + if not isinstance(prior_plan, dict) or prior_plan != placement_plan: + save_cnpg_placement_plan(plan_path, placement_plan) + + glob = self.prole_cfg_data.setdefault("Global", {}) + glob["CNPG_PLACEMENT_PLAN_FILE"] = str(plan_path) + glob["CNPG_PLACEMENT_PLAN_ID"] = str(placement_plan.get("plan_id") or "") + glob["CNPG_PLACEMENT_PLAN_HASH"] = str(placement_plan.get("plan_hash") or "") + + return placement_plan, plan_path + # --------------------------------------------- authority / repair def _authority_context_missing(self) -> bool: enabled = self._get_input_bool( @@ -3742,6 +3889,17 @@ class ProleConsoleInstaller(ProleInstaller): glob = self.prole_cfg_data.setdefault("Global", {}) + cnpg_eligible_nodes = sorted( + { + str((details or {}).get("host") or "").strip() + for details in mounts.values() + if isinstance(details, dict) + and str((details or {}).get("host") or "").strip() + } + ) + if cnpg_eligible_nodes: + glob["CNPG_ELIGIBLE_NODES"] = ",".join(cnpg_eligible_nodes) + # CNPG protected storage defaults (k3s local PV) d001 = mounts.get("d001") or {} d001_base = (d001.get("path") or "").strip() diff --git a/knoe/core/cnpg_placement.py b/knoe/core/cnpg_placement.py new file mode 100644 index 0000000..2db5100 --- /dev/null +++ b/knoe/core/cnpg_placement.py @@ -0,0 +1,186 @@ +from __future__ import annotations + +import hashlib +import json +from pathlib import Path +from typing import Iterable + + +CNPG_PLACEMENT_SCHEMA_VERSION = "v1" + + +def _coerce_positive_int(raw: object, default: int) -> int: + try: + val = int(str(raw).strip()) + if val < 1: + return default + return val + except Exception: + return default + + +def _normalize_nodes(nodes: Iterable[str] | None) -> list[str]: + normalized = { + str(node or "").strip() + for node in (nodes or []) + if str(node or "").strip() + } + return sorted(normalized) + + +def _normalize_assignments(raw: object) -> dict[int, str]: + if not isinstance(raw, dict): + return {} + out: dict[int, str] = {} + for k, v in raw.items(): + try: + ordinal = int(str(k).strip()) + except Exception: + continue + if ordinal < 0: + continue + node = str(v or "").strip() + if not node: + continue + out[ordinal] = node + return out + + +def _assign_round_robin( + *, desired_instances: int, eligible_nodes: list[str], base: dict[int, str] | None = None +) -> dict[int, str]: + assignments = dict(base or {}) + if desired_instances < 1 or not eligible_nodes: + return {} + + for ordinal in range(desired_instances): + if ordinal in assignments: + continue + assignments[ordinal] = eligible_nodes[ordinal % len(eligible_nodes)] + + return {k: assignments[k] for k in sorted(assignments.keys()) if k < desired_instances} + + +def _plan_hash(payload: dict) -> str: + canonical = json.dumps(payload, separators=(",", ":"), sort_keys=True) + digest = hashlib.sha256(canonical.encode("utf-8")).hexdigest() + return digest[:16] + + +def _normalize_plan_for_hash( + *, + cluster_name: str, + desired_instances: int, + eligible_nodes: list[str], + assignments: dict[int, str], +) -> dict: + return { + "schema_version": CNPG_PLACEMENT_SCHEMA_VERSION, + "cluster_name": cluster_name, + "desired_instances": desired_instances, + "eligible_nodes": eligible_nodes, + "assignments": {str(k): assignments[k] for k in sorted(assignments.keys())}, + } + + +def _prior_plan_matches_cluster(prior_plan: dict | None, cluster_name: str) -> bool: + if not isinstance(prior_plan, dict): + return False + return str(prior_plan.get("cluster_name") or "").strip() == cluster_name + + +def _prior_plan_is_eligible(prior_assignments: dict[int, str], eligible_nodes: list[str]) -> bool: + if not prior_assignments: + return False + if not eligible_nodes: + return False + eligible_set = set(eligible_nodes) + return all(node in eligible_set for node in prior_assignments.values()) + + +def plan_cnpg_placement( + *, + cluster_name: str, + desired_instances: int, + candidate_nodes: Iterable[str] | None, + prior_plan: dict | None = None, + rebalance: bool = False, +) -> dict: + cluster = str(cluster_name or "").strip() or "knoe-db" + desired = _coerce_positive_int(desired_instances, 1) + eligible_nodes = _normalize_nodes(candidate_nodes) + + prior_assignments = _normalize_assignments( + (prior_plan or {}).get("assignments") if isinstance(prior_plan, dict) else {} + ) + prior_matches = _prior_plan_matches_cluster(prior_plan, cluster) + prior_eligible = _prior_plan_is_eligible(prior_assignments, eligible_nodes) + + reused = False + regenerated = True + reason = "new_cluster" + + if rebalance: + assignments = _assign_round_robin( + desired_instances=desired, + eligible_nodes=eligible_nodes, + ) + reason = "rebalance_requested" + elif prior_matches and prior_eligible: + # Reuse existing ordinal bindings and only extend/trim as needed. + assignments = _assign_round_robin( + desired_instances=desired, + eligible_nodes=eligible_nodes, + base=prior_assignments, + ) + reused = True + regenerated = False + reason = "reused" + elif prior_matches and prior_assignments and not prior_eligible: + assignments = _assign_round_robin( + desired_instances=desired, + eligible_nodes=eligible_nodes, + ) + reason = "assigned_node_no_longer_eligible" + else: + assignments = _assign_round_robin( + desired_instances=desired, + eligible_nodes=eligible_nodes, + ) + + payload = _normalize_plan_for_hash( + cluster_name=cluster, + desired_instances=desired, + eligible_nodes=eligible_nodes, + assignments=assignments, + ) + plan_hash = _plan_hash(payload) + + return { + **payload, + "plan_id": f"cnpg-placement-{plan_hash}", + "plan_hash": plan_hash, + "metadata": { + "reused": reused, + "regenerated": regenerated, + "reason": reason, + "prior_plan_present": isinstance(prior_plan, dict), + }, + } + + +def load_cnpg_placement_plan(path: Path) -> dict | None: + try: + if not path.exists(): + return None + payload = json.loads(path.read_text()) + except Exception: + return None + if not isinstance(payload, dict): + return None + return payload + + +def save_cnpg_placement_plan(path: Path, plan: dict) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(plan, indent=2, sort_keys=True) + "\n") diff --git a/tests/installer/test_actions_helpers.py b/tests/installer/test_actions_helpers.py index 4719166..22a7a7d 100644 --- a/tests/installer/test_actions_helpers.py +++ b/tests/installer/test_actions_helpers.py @@ -549,6 +549,25 @@ class TestScriptEnvForNamespace: # OpenTofu bootstraps admin password from DB_PASSWORD when unset. assert env.get("OPENTOFU_ADMIN_PASSWORD") == "root-master-password" + def test_exports_cnpg_planner_rendered_env_for_shell_consumption(self, tmp_path): + inst = _TestableInstaller( + inputs={ + "env_setup.PROLE_CONF": str(tmp_path / "conf"), + "init_password.db_namespace": "knoe-system", + }, + project_root=tmp_path, + ) + inst.prole_cfg_data.setdefault("Global", {})["CNPG_ELIGIBLE_NODES"] = "db-b,db-a" + env = inst._script_env_for_namespace("knoe-system") + + plan_file = Path(env.get("CNPG_PLACEMENT_PLAN_FILE", "")) + assert plan_file.exists() + assert env.get("CNPG_PLACEMENT_ELIGIBLE_NODES") == "db-a,db-b" + assert env.get("CNPG_PLACEMENT_PLAN_ID", "").startswith("cnpg-placement-") + # Existing shell flow can consume compatibility selector/instance inputs. + assert env.get("CNPG_CLUSTER_NAME") == "knoe-db" + assert env.get("CNPG_INSTANCES") == "3" + # --------------------------------------------------------------------------- # ensure_db_k8s_secrets diff --git a/tests/installer/test_cnpg_placement.py b/tests/installer/test_cnpg_placement.py new file mode 100644 index 0000000..3083793 --- /dev/null +++ b/tests/installer/test_cnpg_placement.py @@ -0,0 +1,107 @@ +from __future__ import annotations + +from knoe.core.cnpg_placement import ( + load_cnpg_placement_plan, + plan_cnpg_placement, + save_cnpg_placement_plan, +) + + +class TestCnpgPlacementPlanner: + def test_deterministic_eligible_node_ordering_and_round_robin(self): + plan = plan_cnpg_placement( + cluster_name="knoe-db", + desired_instances=5, + candidate_nodes=[" db-b ", "db-a", "db-b", "db-c"], + ) + assert plan["eligible_nodes"] == ["db-a", "db-b", "db-c"] + assert plan["assignments"] == { + "0": "db-a", + "1": "db-b", + "2": "db-c", + "3": "db-a", + "4": "db-b", + } + + def test_reuses_persisted_plan_when_still_valid(self): + prior = plan_cnpg_placement( + cluster_name="knoe-db", + desired_instances=3, + candidate_nodes=["db-a", "db-b"], + ) + plan = plan_cnpg_placement( + cluster_name="knoe-db", + desired_instances=3, + candidate_nodes=["db-b", "db-a", "db-c"], + prior_plan=prior, + ) + assert plan["metadata"]["reused"] is True + assert plan["metadata"]["regenerated"] is False + # New eligible node must not reshuffle existing ordinals. + assert plan["assignments"] == prior["assignments"] + + def test_regenerates_when_assigned_node_becomes_ineligible(self): + prior = { + "cluster_name": "knoe-db", + "assignments": {"0": "db-a", "1": "db-b", "2": "db-a"}, + } + plan = plan_cnpg_placement( + cluster_name="knoe-db", + desired_instances=3, + candidate_nodes=["db-a", "db-c"], + prior_plan=prior, + ) + assert plan["metadata"]["regenerated"] is True + assert plan["metadata"]["reason"] == "assigned_node_no_longer_eligible" + assert set(plan["assignments"].values()) <= {"db-a", "db-c"} + + def test_rebalance_regenerates_even_with_valid_prior(self): + prior = { + "cluster_name": "knoe-db", + "assignments": {"0": "db-a", "1": "db-a", "2": "db-a"}, + } + plan = plan_cnpg_placement( + cluster_name="knoe-db", + desired_instances=3, + candidate_nodes=["db-a", "db-b", "db-c"], + prior_plan=prior, + rebalance=True, + ) + assert plan["metadata"]["regenerated"] is True + assert plan["metadata"]["reason"] == "rebalance_requested" + assert plan["assignments"] == {"0": "db-a", "1": "db-b", "2": "db-c"} + + def test_graceful_when_fewer_eligible_nodes_than_instances(self): + plan = plan_cnpg_placement( + cluster_name="knoe-db", + desired_instances=4, + candidate_nodes=["db-a"], + ) + assert plan["assignments"] == { + "0": "db-a", + "1": "db-a", + "2": "db-a", + "3": "db-a", + } + + +class TestCnpgPlacementPersistence: + def test_persisted_plan_can_be_reloaded_and_reused(self, tmp_path): + plan_path = tmp_path / "cnpg-placement" / "knoe-system-knoe-db.json" + initial = plan_cnpg_placement( + cluster_name="knoe-db", + desired_instances=3, + candidate_nodes=["db-a", "db-b"], + ) + save_cnpg_placement_plan(plan_path, initial) + + prior = load_cnpg_placement_plan(plan_path) + assert prior is not None + reused = plan_cnpg_placement( + cluster_name="knoe-db", + desired_instances=3, + candidate_nodes=["db-a", "db-b", "db-c"], + prior_plan=prior, + ) + assert reused["metadata"]["reused"] is True + assert reused["assignments"] == initial["assignments"]