tegwick
f18eb0b8df
MAS-WP-0007 is finished. power.sh scales the MX and statute together and records the power annotation. apply.sh keeps an off facility off. The live facility was annotated power=on and was not restarted. Assistant: grok Assistant-Session: 01a198fa-fd27-7e41-a0a2-57355c3e4efd
81 lines
2.5 KiB
Python
81 lines
2.5 KiB
Python
"""Render the MAS facility manifest for a power state.
|
|
|
|
This file does not talk to the cluster. power.sh and apply.sh do.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
|
|
SWITCHABLE = ("mas-mx", "mas-statute")
|
|
|
|
|
|
def render(manifest: str, image: str, power: str) -> str:
|
|
"""Return the manifest with IMAGE filled in and replicas set from power.
|
|
|
|
`on` keeps one replica of each switchable Deployment. `off` sets both
|
|
to zero. Services, the evidence volume, the Ingress, and the
|
|
ExternalSecret are copied through unchanged.
|
|
"""
|
|
if power not in ("on", "off"):
|
|
raise ValueError(f"power must be on or off, got {power!r}")
|
|
if not image or any(ch.isspace() for ch in image):
|
|
raise ValueError("image must be a single token")
|
|
if "image: IMAGE" not in manifest:
|
|
raise ValueError("manifest has no image: IMAGE placeholder")
|
|
|
|
rendered = manifest.replace("image: IMAGE", f"image: {image}")
|
|
docs = rendered.split("\n---\n")
|
|
seen: set[str] = set()
|
|
out: list[str] = []
|
|
for doc in docs:
|
|
name = _deployment_name(doc)
|
|
if name in SWITCHABLE:
|
|
seen.add(name)
|
|
doc = _set_replicas(doc, 0 if power == "off" else 1)
|
|
out.append(doc)
|
|
missing = [name for name in SWITCHABLE if name not in seen]
|
|
if missing:
|
|
raise ValueError("manifest is missing Deployments: " + ", ".join(missing))
|
|
return "\n---\n".join(out)
|
|
|
|
|
|
def _deployment_name(doc: str) -> str | None:
|
|
kind = None
|
|
name = None
|
|
for line in doc.splitlines():
|
|
if line.startswith("kind:"):
|
|
kind = line.split(":", 1)[1].strip()
|
|
elif line.startswith(" name:") and name is None:
|
|
name = line.split(":", 1)[1].strip()
|
|
if kind == "Deployment":
|
|
return name
|
|
return None
|
|
|
|
|
|
def _set_replicas(doc: str, count: int) -> str:
|
|
lines = doc.splitlines(keepends=True)
|
|
found = 0
|
|
out = []
|
|
for line in lines:
|
|
if line.startswith(" replicas:"):
|
|
found += 1
|
|
newline = "\n" if line.endswith("\n") else ""
|
|
out.append(f" replicas: {count}{newline}")
|
|
else:
|
|
out.append(line)
|
|
if found != 1:
|
|
raise ValueError(f"expected one replicas field, found {found}")
|
|
return "".join(out)
|
|
|
|
|
|
def main(argv: list[str]) -> int:
|
|
if len(argv) != 4 or argv[1] != "render":
|
|
print("usage: facility.py render IMAGE on|off", file=sys.stderr)
|
|
return 2
|
|
sys.stdout.write(render(sys.stdin.read(), argv[2], argv[3]))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main(sys.argv))
|