chrisfu/prole
1
0
Fork 0
mirror of https://github.com/dredx/prole.git synced 2026-09-23 12:03:59 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Add CNPG placement planner with persistence and integration tests

Browse Source 
- 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.
This commit is contained in:
chrisfu 2026-03-23 00:06:47 -07:00
parent d1e2a8273d
commit 59c8ea3f8f
5 changed files with 593 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

124
etc/init_cloudnative_pg.sh
View File

@ -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 \

158
knoe/core/actions.py
View File

@ -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()

186
knoe/core/cnpg_placement.py Normal file
View File

@ -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")

19
tests/installer/test_actions_helpers.py
View File

@ -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

107
tests/installer/test_cnpg_placement.py Normal file
View File

@ -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"]