Add hostname parsers + find_drift() to capacity-scan.py

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
sjat 2026-06-01 10:24:11 +02:00
parent b240fa8bfe
commit 8ed00c9206
2 changed files with 52 additions and 0 deletions

View file

@ -78,3 +78,33 @@ def compute_rollup(node_rows, workload_rows):
or node["alloc_disk_gb"] > node["disk_gb"]
)
return nodes
def parse_tf_hostnames(tf_json):
"""Hostnames from `terraform output -json` (the `vms` map keys)."""
data = json.loads(tf_json)
return set(data.get("vms", {}).get("value", {}).keys())
def parse_inventory_hostnames(inv_json):
"""Hostnames from `ansible-inventory --list` (_meta.hostvars keys)."""
data = json.loads(inv_json)
return set(data.get("_meta", {}).get("hostvars", {}).keys())
def find_drift(workload_rows, known_hostnames):
"""Warn when reference.md workloads and live hostnames disagree. Silent when
no hostnames are known (pre-provisioning) nothing to compare against."""
warnings = []
declared = {w["workload"] for w in workload_rows}
if not known_hostnames:
return warnings
for name in sorted(declared - known_hostnames):
warnings.append(
f"reference.md lists '{name}' but no Terraform/inventory host declares it"
)
for name in sorted(known_hostnames - declared):
warnings.append(
f"host '{name}' exists in Terraform/inventory but is absent from reference.md"
)
return warnings

View file

@ -58,3 +58,25 @@ def test_compute_rollup_ignores_workloads_on_unknown_nodes():
[{"workload": "ghost", "node": "nope", "cores": "1", "ram_mb": "512", "disk_gb": "10"}],
)
assert nodes["pve0"]["alloc_cores"] == 0
def test_parse_tf_hostnames_reads_vms_value_keys():
tf_json = '{"vms": {"value": {"dns1": {"ip": "10.20.0.10", "group": "docker_hosts"}}}}'
assert cs.parse_tf_hostnames(tf_json) == {"dns1"}
def test_parse_inventory_hostnames_reads_meta_hostvars():
inv_json = '{"_meta": {"hostvars": {"dns1": {}, "proxy": {}}}}'
assert cs.parse_inventory_hostnames(inv_json) == {"dns1", "proxy"}
def test_find_drift_reports_both_directions():
workload_rows = [{"workload": "dns1", "node": "pve0", "cores": "1", "ram_mb": "512", "disk_gb": "10"}]
warnings = cs.find_drift(workload_rows, {"proxy"})
assert any("dns1" in w and "no Terraform" in w for w in warnings)
assert any("proxy" in w and "absent from reference.md" in w for w in warnings)
def test_find_drift_silent_when_no_hostnames_known():
workload_rows = [{"workload": "dns1", "node": "pve0", "cores": "1", "ram_mb": "512", "disk_gb": "10"}]
assert cs.find_drift(workload_rows, set()) == []