Add ServiceDefinition renderer (T-0.3, R-GITOPS-02)
catalog/render.py compiles a ServiceDefinition into the manifest set defined by schemas/renderer-contract.md: Namespace + default-deny baseline (R-NET-01), Deployment/Service or KubeVirt VM + CDI DataVolume (R-PLAT-02/03), PVCs (R-STATE), per-dependency CiliumNetworkPolicy allows (R-NET-02), egress-class steering label (R-NET-05), HTTPRoutes on the shared Gateway (R-TLS/R-DNS), ServiceMonitor + SLI alerts (R-OBS-01), Flux image-automation (R-UPD), auth-mode/backup hints (R-IDP). Deterministic (golden-diffable) with a built-in --selftest; single dependency (PyYAML) so the conformance harness stays stdlib-only. Documents the mechanism choice in catalog/README.md and the script header. Co-Authored-By: Claude Opus 4.8 <[email protected]>
This commit is contained in:
jakman85andClaude Opus 4.8 committed
1 parent
f49b8c2e02
commit
6b2c04ed5c
2 files changed
+800
No files matched your search
@@ -0,0 +1,774 @@
|
||||
#!/usr/bin/env python3
|
||||
"""ServiceDefinition renderer (task T-0.3) — the platform's core GitOps contract.
|
||||
|
||||
Consumes ONE ServiceDefinition (schemas/servicedefinition.schema.json, catalog/*.yaml)
|
||||
and emits the concrete manifest set the renderer contract specifies
|
||||
(schemas/renderer-contract.md): Namespace + default-deny baseline, Deployment/Service or
|
||||
KubeVirt VirtualMachine + CDI DataVolume, PVCs, per-dependency CiliumNetworkPolicy allows,
|
||||
egress-class steering label, HTTPRoute(s), ServiceMonitor/alerts, Flux image-automation,
|
||||
and the auth-mode / backup wiring hints.
|
||||
|
||||
Requirements: R-GITOPS-02 (single reviewable object wires every requirement), R-NET-01/02
|
||||
(default-deny + explicit allows), R-NET-05 (egress class), R-STATE-01/02/04 (storage),
|
||||
R-TLS/R-DNS (ingressRoutes/exposure), R-OBS-01 (observability), R-UPD-01/02/03 (update),
|
||||
R-IDP-02/06 (auth), R-PLAT-02/03 (kind/source).
|
||||
|
||||
Design (documented per the task's "if the contract is ambiguous, choose the simplest
|
||||
reproducible option"): a deterministic Python renderer. The contract does not mandate a
|
||||
mechanism (script / kustomize generator / templates); a render-then-consume script is the
|
||||
simplest reproducible option the repo can run in CI and that Flux consumes from a rendered
|
||||
output path. CI invokes it via `render.py <sd>` (or `--all`); the offline L1 render
|
||||
conformance verifier diffs its deterministic output against committed goldens.
|
||||
|
||||
Determinism: fixed document order + fixed key order + PyYAML with sort_keys=False and a
|
||||
wide line width, so identical input always yields byte-identical output (golden-diffable).
|
||||
|
||||
Dependency: PyYAML (input parse + output emit). This is the ONE renderer dependency; the
|
||||
conformance harness stays stdlib-only and only shells out to this script + diffs text.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
try:
|
||||
import yaml
|
||||
except ImportError: # pragma: no cover - explicit, actionable failure
|
||||
sys.stderr.write(
|
||||
"render.py requires PyYAML (the renderer's single dependency). "
|
||||
"Install with: pip install pyyaml\n"
|
||||
)
|
||||
raise SystemExit(2)
|
||||
|
||||
# --- Label / naming conventions (match platform/kyverno/policies + platform/*/) --------
|
||||
LBL = "homelab.keithjackson.dev"
|
||||
ZONE = "home.keithjackson.dev"
|
||||
GATEWAY_NAME = "shared"
|
||||
GATEWAY_NS = "gateway"
|
||||
KP_RELEASE = "kube-prometheus-stack" # ServiceMonitor/PrometheusRule discovery label
|
||||
|
||||
# egress classes and auth modes fixed by docs/03 / docs/13 + the Kyverno enums.
|
||||
EGRESS_CLASSES = ("internet", "vpn")
|
||||
AUTH_MODES = ("oidc", "forward-auth", "native", "none")
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Defaults (renderer-contract.md "Default posture") — materialize BEFORE mapping.
|
||||
# =============================================================================
|
||||
def apply_defaults(sd: dict) -> dict:
|
||||
"""Return a copy of the ServiceDefinition with the contract's default posture filled.
|
||||
|
||||
A field absent from a ServiceDefinition renders from its default posture, not "not at
|
||||
all" (renderer-contract.md).
|
||||
"""
|
||||
s = dict(sd)
|
||||
s.setdefault("kind", "container")
|
||||
s.setdefault("exposure", "internal")
|
||||
s.setdefault("egress", "internet")
|
||||
s.setdefault("storage", [])
|
||||
s.setdefault("dependencies", [])
|
||||
s.setdefault("ingressRoutes", [])
|
||||
|
||||
# auth.mode default posture: forward-auth for any exposed UI, else none (internal, no
|
||||
# UI). An explicit auth block always wins (R-IDP-02).
|
||||
auth = dict(s.get("auth") or {})
|
||||
if "mode" not in auth:
|
||||
auth["mode"] = "forward-auth" if s["ingressRoutes"] else "none"
|
||||
auth.setdefault("groups", [])
|
||||
s["auth"] = auth
|
||||
|
||||
update = dict(s.get("update") or {})
|
||||
update.setdefault("channel", "stable")
|
||||
update.setdefault("autoRollbackOnFailedHealth", True)
|
||||
s["update"] = update
|
||||
|
||||
obs = dict(s.get("observability") or {})
|
||||
obs.setdefault("metrics", True)
|
||||
obs.setdefault("slis", [])
|
||||
s["observability"] = obs
|
||||
|
||||
# storage entry class default is longhorn (replicated tier).
|
||||
s["storage"] = [{**{"class": "longhorn"}, **v} for v in s["storage"]]
|
||||
# dependency protocol default is tcp.
|
||||
s["dependencies"] = [{**{"protocol": "tcp"}, **d} for d in s["dependencies"]]
|
||||
return s
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Field -> resource mapping (renderer-contract.md "Field -> resource mapping")
|
||||
# =============================================================================
|
||||
def r_namespace(s: dict) -> dict:
|
||||
"""Namespace = the service trust zone / isolation boundary (R-NET, docs/03).
|
||||
|
||||
Carries app-namespace=true so Kyverno generates the default-deny baseline and the
|
||||
egress-class / auth-mode admission policies apply. gateway-access=shared lets its
|
||||
HTTPRoutes attach to the shared Gateway (only when it exposes routes).
|
||||
"""
|
||||
labels = {f"{LBL}/app-namespace": "true"}
|
||||
if s["ingressRoutes"]:
|
||||
labels["gateway-access"] = "shared"
|
||||
ann = {}
|
||||
backup = (s.get("backup") or {}).get("policy")
|
||||
if backup:
|
||||
ann[f"{LBL}/backup-policy"] = backup # R-STATE-03 off-cluster schedule ref (hint)
|
||||
meta = {"name": s["name"], "labels": labels}
|
||||
if ann:
|
||||
meta["annotations"] = ann
|
||||
return {"apiVersion": "v1", "kind": "Namespace", "metadata": meta}
|
||||
|
||||
|
||||
def r_default_deny(s: dict) -> dict:
|
||||
"""The default-deny baseline: deny all ingress+egress, allow ONLY DNS egress (R-NET-01).
|
||||
|
||||
Rendered explicitly (GitOps at namespace creation, renderer-contract §dependencies) —
|
||||
it matches the Kyverno-generated baseline shape (platform/kyverno/.../generated-default
|
||||
-deny.yaml) so the render carries the guarantee even before admission generates it.
|
||||
"""
|
||||
return {
|
||||
"apiVersion": "networking.k8s.io/v1",
|
||||
"kind": "NetworkPolicy",
|
||||
"metadata": {
|
||||
"name": "default-deny",
|
||||
"namespace": s["name"],
|
||||
"labels": {f"{LBL}/baseline": "default-deny"},
|
||||
},
|
||||
"spec": {
|
||||
"podSelector": {},
|
||||
"policyTypes": ["Ingress", "Egress"],
|
||||
"ingress": [],
|
||||
"egress": [
|
||||
{
|
||||
"to": [
|
||||
{
|
||||
"namespaceSelector": {
|
||||
"matchLabels": {
|
||||
"kubernetes.io/metadata.name": "kube-system"
|
||||
}
|
||||
}
|
||||
}
|
||||
],
|
||||
"ports": [
|
||||
{"protocol": "UDP", "port": 53},
|
||||
{"protocol": "TCP", "port": 53},
|
||||
],
|
||||
}
|
||||
],
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _pod_labels(s: dict) -> dict:
|
||||
"""Workload pod labels: identity + the egress-class steering hook + auth-mode.
|
||||
|
||||
egress-class is what the shared Cilium Egress Gateway selects on to steer vpn-class
|
||||
pods (enforcement is OUTSIDE the workload — renderer-contract §egress; ADR-0015);
|
||||
auth-mode satisfies the Kyverno require-auth-mode-label policy (R-IDP-02).
|
||||
"""
|
||||
return {
|
||||
"app.kubernetes.io/name": s["name"],
|
||||
f"{LBL}/egress-class": s["egress"],
|
||||
f"{LBL}/auth-mode": s["auth"]["mode"],
|
||||
}
|
||||
|
||||
|
||||
def _service_ports(s: dict) -> list:
|
||||
ports = sorted({r["port"] for r in s["ingressRoutes"]})
|
||||
return [{"name": f"port-{p}", "port": p, "targetPort": p} for p in ports]
|
||||
|
||||
|
||||
def r_pvcs(s: dict) -> list:
|
||||
"""One PVC per storage entry (R-STATE-01/02/04).
|
||||
|
||||
longhorn -> 3-replica replicated tier; bulk -> external media tier (not 3x replicated).
|
||||
"""
|
||||
out = []
|
||||
for vol in s["storage"]:
|
||||
cls = vol.get("class", "longhorn")
|
||||
storage_class = "longhorn" if cls == "longhorn" else "bulk"
|
||||
spec = {
|
||||
"accessModes": ["ReadWriteOnce"],
|
||||
"storageClassName": storage_class,
|
||||
}
|
||||
if vol.get("size"):
|
||||
spec["resources"] = {"requests": {"storage": vol["size"]}}
|
||||
out.append(
|
||||
{
|
||||
"apiVersion": "v1",
|
||||
"kind": "PersistentVolumeClaim",
|
||||
"metadata": {
|
||||
"name": f"{s['name']}-{vol['name']}",
|
||||
"namespace": s["name"],
|
||||
"labels": {"app.kubernetes.io/name": s["name"]},
|
||||
},
|
||||
"spec": spec,
|
||||
}
|
||||
)
|
||||
return out
|
||||
|
||||
|
||||
def r_deployment(s: dict) -> dict:
|
||||
"""Deployment for a container workload (R-PLAT-02)."""
|
||||
pod_labels = _pod_labels(s)
|
||||
container = {
|
||||
"name": s["name"],
|
||||
"image": s["source"]["image"],
|
||||
}
|
||||
ports = _service_ports(s)
|
||||
if ports:
|
||||
container["ports"] = [{"containerPort": p["port"]} for p in ports]
|
||||
|
||||
volume_mounts, volumes = [], []
|
||||
for vol in s["storage"]:
|
||||
volume_mounts.append({"name": vol["name"], "mountPath": f"/data/{vol['name']}"})
|
||||
volumes.append(
|
||||
{
|
||||
"name": vol["name"],
|
||||
"persistentVolumeClaim": {"claimName": f"{s['name']}-{vol['name']}"},
|
||||
}
|
||||
)
|
||||
if volume_mounts:
|
||||
container["volumeMounts"] = volume_mounts
|
||||
|
||||
pod_spec = {"containers": [container]}
|
||||
if volumes:
|
||||
pod_spec["volumes"] = volumes
|
||||
|
||||
ann = _auth_annotations(s)
|
||||
tmpl_meta = {"labels": pod_labels}
|
||||
if ann:
|
||||
tmpl_meta["annotations"] = ann
|
||||
|
||||
return {
|
||||
"apiVersion": "apps/v1",
|
||||
"kind": "Deployment",
|
||||
"metadata": {
|
||||
"name": s["name"],
|
||||
"namespace": s["name"],
|
||||
"labels": {"app.kubernetes.io/name": s["name"]},
|
||||
},
|
||||
"spec": {
|
||||
"replicas": 1,
|
||||
"selector": {"matchLabels": {"app.kubernetes.io/name": s["name"]}},
|
||||
"template": {"metadata": tmpl_meta, "spec": pod_spec},
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def r_service(s: dict):
|
||||
ports = _service_ports(s)
|
||||
if not ports:
|
||||
return None
|
||||
return {
|
||||
"apiVersion": "v1",
|
||||
"kind": "Service",
|
||||
"metadata": {
|
||||
"name": s["name"],
|
||||
"namespace": s["name"],
|
||||
"labels": {"app.kubernetes.io/name": s["name"]},
|
||||
},
|
||||
"spec": {
|
||||
"selector": {"app.kubernetes.io/name": s["name"]},
|
||||
"ports": ports,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def r_dependency_policies(s: dict) -> list:
|
||||
"""One CiliumNetworkPolicy egress allow per dependency, on top of default-deny (R-NET-02).
|
||||
|
||||
Everything undeclared stays denied; empty dependencies => reaches only DNS.
|
||||
"""
|
||||
out = []
|
||||
for dep in s["dependencies"]:
|
||||
proto = dep.get("protocol", "tcp").upper()
|
||||
out.append(
|
||||
{
|
||||
"apiVersion": "cilium.io/v2",
|
||||
"kind": "CiliumNetworkPolicy",
|
||||
"metadata": {
|
||||
"name": f"{s['name']}-allow-{dep['to']}-{dep['port']}",
|
||||
"namespace": s["name"],
|
||||
"labels": {"app.kubernetes.io/name": s["name"]},
|
||||
},
|
||||
"spec": {
|
||||
"endpointSelector": {
|
||||
"matchLabels": {"app.kubernetes.io/name": s["name"]}
|
||||
},
|
||||
"egress": [
|
||||
{
|
||||
"toEndpoints": [
|
||||
{"matchLabels": {"app.kubernetes.io/name": dep["to"]}}
|
||||
],
|
||||
"toPorts": [
|
||||
{
|
||||
"ports": [
|
||||
{"port": str(dep["port"]), "protocol": proto}
|
||||
]
|
||||
}
|
||||
],
|
||||
}
|
||||
],
|
||||
},
|
||||
}
|
||||
)
|
||||
return out
|
||||
|
||||
|
||||
def _route_meta(s: dict, route: dict) -> dict:
|
||||
meta = {
|
||||
"name": f"{s['name']}-{route['port']}",
|
||||
"namespace": s["name"],
|
||||
"labels": {"app.kubernetes.io/name": s["name"]},
|
||||
}
|
||||
ann = _auth_annotations(s)
|
||||
if ann:
|
||||
meta["annotations"] = ann
|
||||
return meta
|
||||
|
||||
|
||||
def r_httproutes(s: dict) -> list:
|
||||
"""One HTTPRoute per ingressRoute, attaching host -> service port on the shared Gateway.
|
||||
|
||||
TLS is the Gateway's wildcard *.home.keithjackson.dev cert (no per-service cert step);
|
||||
ExternalDNS manages the record from the HTTPRoute (R-TLS/R-DNS, renderer-contract).
|
||||
"""
|
||||
out = []
|
||||
for route in s["ingressRoutes"]:
|
||||
out.append(
|
||||
{
|
||||
"apiVersion": "gateway.networking.k8s.io/v1",
|
||||
"kind": "HTTPRoute",
|
||||
"metadata": _route_meta(s, route),
|
||||
"spec": {
|
||||
"parentRefs": [
|
||||
{
|
||||
"name": GATEWAY_NAME,
|
||||
"namespace": GATEWAY_NS,
|
||||
"sectionName": "https",
|
||||
}
|
||||
],
|
||||
"hostnames": [route["host"]],
|
||||
"rules": [
|
||||
{
|
||||
"matches": [
|
||||
{"path": {"type": "PathPrefix", "value": "/"}}
|
||||
],
|
||||
"backendRefs": [
|
||||
{"name": s["name"], "port": route["port"]}
|
||||
],
|
||||
}
|
||||
],
|
||||
},
|
||||
}
|
||||
)
|
||||
return out
|
||||
|
||||
|
||||
def _auth_annotations(s: dict) -> dict:
|
||||
"""Auth wiring hints (renderer-contract §auth): the platform (oauth2-proxy/Keycloak)
|
||||
consumes these to protect ingressRoutes. oidc/forward-auth carry the authorized groups
|
||||
(R-IDP-06: access is 'in an authorized group', not merely 'logged in')."""
|
||||
mode = s["auth"]["mode"]
|
||||
if mode in ("oidc", "forward-auth"):
|
||||
groups = ",".join(s["auth"].get("groups", []))
|
||||
return {
|
||||
f"{LBL}/auth-mode": mode,
|
||||
f"{LBL}/auth-groups": groups,
|
||||
}
|
||||
return {}
|
||||
|
||||
|
||||
def r_servicemonitor(s: dict):
|
||||
if not s["observability"]["metrics"]:
|
||||
return None
|
||||
return {
|
||||
"apiVersion": "monitoring.coreos.com/v1",
|
||||
"kind": "ServiceMonitor",
|
||||
"metadata": {
|
||||
"name": s["name"],
|
||||
"namespace": s["name"],
|
||||
"labels": {"release": KP_RELEASE, "app.kubernetes.io/name": s["name"]},
|
||||
},
|
||||
"spec": {
|
||||
"selector": {"matchLabels": {"app.kubernetes.io/name": s["name"]}},
|
||||
"namespaceSelector": {"matchNames": [s["name"]]},
|
||||
"endpoints": [{"port": "metrics", "interval": "30s"}],
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def r_prometheusrule(s: dict):
|
||||
"""Each SLI becomes at least one alert rule (R-OBS-01)."""
|
||||
slis = s["observability"]["slis"]
|
||||
if not slis:
|
||||
return None
|
||||
rules = []
|
||||
for sli in slis:
|
||||
if sli == "up":
|
||||
rules.append(
|
||||
{
|
||||
"alert": f"{s['name']}-down".replace("_", "-"),
|
||||
"expr": f'up{{job="{s["name"]}"}} == 0',
|
||||
"for": "5m",
|
||||
"labels": {"severity": "critical", "service": s["name"]},
|
||||
"annotations": {
|
||||
"summary": f"{s['name']} is down",
|
||||
"description": f"{s['name']} has been unreachable for 5m (SLI: up).",
|
||||
},
|
||||
}
|
||||
)
|
||||
else:
|
||||
rules.append(
|
||||
{
|
||||
"alert": f"{s['name']}-{sli}".replace("_", "-"),
|
||||
"expr": f'absent({sli}{{service="{s["name"]}"}})',
|
||||
"for": "10m",
|
||||
"labels": {"severity": "warning", "service": s["name"]},
|
||||
"annotations": {"summary": f"{s['name']} SLI {sli} unhealthy"},
|
||||
}
|
||||
)
|
||||
return {
|
||||
"apiVersion": "monitoring.coreos.com/v1",
|
||||
"kind": "PrometheusRule",
|
||||
"metadata": {
|
||||
"name": f"{s['name']}-slis",
|
||||
"namespace": s["name"],
|
||||
"labels": {"release": KP_RELEASE, "app.kubernetes.io/name": s["name"]},
|
||||
},
|
||||
"spec": {"groups": [{"name": f"{s['name']}.slis", "rules": rules}]},
|
||||
}
|
||||
|
||||
|
||||
def r_image_automation(s: dict) -> list:
|
||||
"""container kind -> Flux ImageRepository + ImagePolicy encoding channel/range (R-UPD).
|
||||
|
||||
Flux image-automation advances the running version on its scan interval; no human/AI
|
||||
edits a tag (R-UPD-02). `pin` freezes automation. vm kind uses in-guest updates instead.
|
||||
"""
|
||||
if s["kind"] != "container":
|
||||
return []
|
||||
image = s["source"]["image"].split(":", 1)[0]
|
||||
upd = s["update"]
|
||||
interval = "1h0m0s"
|
||||
imagerepo = {
|
||||
"apiVersion": "image.toolkit.fluxcd.io/v1beta2",
|
||||
"kind": "ImageRepository",
|
||||
"metadata": {"name": s["name"], "namespace": s["name"]},
|
||||
"spec": {"image": image, "interval": interval},
|
||||
}
|
||||
policy_spec: dict = {"imageRepositoryRef": {"name": s["name"]}}
|
||||
if upd.get("pin"):
|
||||
# Operator freeze escape hatch: pin to a fixed tag, stopping automation.
|
||||
policy_spec["filterTags"] = {"pattern": f"^{upd['pin']}$"}
|
||||
policy_spec["policy"] = {"alphabetical": {"order": "asc"}}
|
||||
elif upd.get("range"):
|
||||
policy_spec["policy"] = {"semver": {"range": upd["range"]}}
|
||||
else:
|
||||
policy_spec["policy"] = {"semver": {"range": ">=0.0.0"}}
|
||||
imagepolicy = {
|
||||
"apiVersion": "image.toolkit.fluxcd.io/v1beta2",
|
||||
"kind": "ImagePolicy",
|
||||
"metadata": {
|
||||
"name": s["name"],
|
||||
"namespace": s["name"],
|
||||
"annotations": {
|
||||
f"{LBL}/update-channel": upd.get("channel", "stable"),
|
||||
f"{LBL}/auto-rollback": str(
|
||||
upd.get("autoRollbackOnFailedHealth", True)
|
||||
).lower(),
|
||||
},
|
||||
},
|
||||
"spec": policy_spec,
|
||||
}
|
||||
return [imagerepo, imagepolicy]
|
||||
|
||||
|
||||
# --- VM (kind: vm) path: KubeVirt VirtualMachine + CDI DataVolume ----------------------
|
||||
def r_datavolume(s: dict) -> dict:
|
||||
"""Standalone CDI DataVolume (NOT a dataVolumeTemplate) so the disk OUTLIVES the VM
|
||||
and survives destroy/rebuild (R-STATE-01) — matches the vm-lifecycle fixture shape."""
|
||||
root = s["storage"][0] if s["storage"] else {"name": "rootdisk", "size": "10Gi"}
|
||||
return {
|
||||
"apiVersion": "cdi.kubevirt.io/v1beta1",
|
||||
"kind": "DataVolume",
|
||||
"metadata": {
|
||||
"name": f"{s['name']}-{root['name']}",
|
||||
"namespace": s["name"],
|
||||
},
|
||||
"spec": {
|
||||
"source": {"registry": {"url": f"docker://{s['source']['vmTemplate']}"}},
|
||||
"storage": {
|
||||
"accessModes": ["ReadWriteOnce"],
|
||||
"storageClassName": "longhorn",
|
||||
"resources": {"requests": {"storage": root.get("size", "10Gi")}},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
VM_CLOUD_INIT = (
|
||||
"#cloud-config\n"
|
||||
"package_update: true\n"
|
||||
"packages:\n"
|
||||
" - dnf-automatic\n"
|
||||
"write_files:\n"
|
||||
" - path: /etc/dnf/automatic.conf\n"
|
||||
" content: |\n"
|
||||
" [commands]\n"
|
||||
" upgrade_type = default\n"
|
||||
" random_sleep = 0\n"
|
||||
" download_updates = yes\n"
|
||||
" apply_updates = yes\n"
|
||||
"runcmd:\n"
|
||||
" - [ systemctl, enable, --now, dnf-automatic.timer ]\n"
|
||||
)
|
||||
|
||||
|
||||
def r_virtualmachine(s: dict) -> dict:
|
||||
"""KubeVirt VirtualMachine; egress-class propagates to the virt-launcher pod so the
|
||||
shared Cilium egress steering + Kyverno policy see the class (R-PLAT-03; ADR-0015).
|
||||
In-guest dnf-automatic delivers auto-updates with no human step (R-UPD-01, docs/11)."""
|
||||
root = s["storage"][0] if s["storage"] else {"name": "rootdisk"}
|
||||
pod_labels = {
|
||||
"kubevirt.io/vm": s["name"],
|
||||
f"{LBL}/egress-class": s["egress"],
|
||||
f"{LBL}/auth-mode": s["auth"]["mode"],
|
||||
}
|
||||
return {
|
||||
"apiVersion": "kubevirt.io/v1",
|
||||
"kind": "VirtualMachine",
|
||||
"metadata": {
|
||||
"name": s["name"],
|
||||
"namespace": s["name"],
|
||||
"labels": {"app.kubernetes.io/name": s["name"]},
|
||||
},
|
||||
"spec": {
|
||||
"running": False,
|
||||
"template": {
|
||||
"metadata": {"labels": pod_labels},
|
||||
"spec": {
|
||||
"domain": {
|
||||
"cpu": {"cores": 1},
|
||||
"memory": {"guest": "1536Mi"},
|
||||
"devices": {
|
||||
"disks": [
|
||||
{"name": "rootdisk", "disk": {"bus": "virtio"}},
|
||||
{"name": "cloudinit", "disk": {"bus": "virtio"}},
|
||||
],
|
||||
"interfaces": [{"name": "default", "masquerade": {}}],
|
||||
},
|
||||
"resources": {"requests": {"memory": "1536Mi"}},
|
||||
},
|
||||
"networks": [{"name": "default", "pod": {}}],
|
||||
"volumes": [
|
||||
{
|
||||
"name": "rootdisk",
|
||||
"dataVolume": {"name": f"{s['name']}-{root['name']}"},
|
||||
},
|
||||
{
|
||||
"name": "cloudinit",
|
||||
"cloudInitNoCloud": {"userData": VM_CLOUD_INIT},
|
||||
},
|
||||
],
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Top-level render + emit
|
||||
# =============================================================================
|
||||
def render(sd: dict) -> list:
|
||||
"""Return an ordered list of (header_comment, resource_dict) for a ServiceDefinition."""
|
||||
s = apply_defaults(sd)
|
||||
if s["kind"] not in ("container", "vm"):
|
||||
raise ValueError(f"unknown kind: {s['kind']!r}")
|
||||
if s["egress"] not in EGRESS_CLASSES:
|
||||
raise ValueError(f"invalid egress class: {s['egress']!r}")
|
||||
if s["auth"]["mode"] not in AUTH_MODES:
|
||||
raise ValueError(f"invalid auth.mode: {s['auth']['mode']!r}")
|
||||
|
||||
docs: list = []
|
||||
docs.append(("Namespace (trust zone; default-deny + policies apply)", r_namespace(s)))
|
||||
docs.append(("Default-deny baseline (R-NET-01): deny all, allow only DNS", r_default_deny(s)))
|
||||
|
||||
if s["kind"] == "container":
|
||||
# Container PVCs. (For a vm, the CDI DataVolume owns/creates the disk PVC, so we
|
||||
# never render a plain PVC for the VM root disk — it would collide with CDI.)
|
||||
for pvc in r_pvcs(s):
|
||||
docs.append((f"PVC {pvc['metadata']['name']} (R-STATE)", pvc))
|
||||
docs.append(("Deployment (R-PLAT-02)", r_deployment(s)))
|
||||
svc = r_service(s)
|
||||
if svc:
|
||||
docs.append(("Service", svc))
|
||||
else:
|
||||
docs.append(("CDI DataVolume (standalone; survives rebuild, R-STATE-01)", r_datavolume(s)))
|
||||
docs.append(("KubeVirt VirtualMachine (R-PLAT-03)", r_virtualmachine(s)))
|
||||
|
||||
for cnp in r_dependency_policies(s):
|
||||
docs.append((f"CiliumNetworkPolicy egress allow (R-NET-02): {cnp['metadata']['name']}", cnp))
|
||||
|
||||
for hr in r_httproutes(s):
|
||||
docs.append((f"HTTPRoute {hr['metadata']['name']} (R-TLS/R-DNS)", hr))
|
||||
|
||||
sm = r_servicemonitor(s)
|
||||
if sm:
|
||||
docs.append(("ServiceMonitor (R-OBS-01)", sm))
|
||||
pr = r_prometheusrule(s)
|
||||
if pr:
|
||||
docs.append(("PrometheusRule SLI alerts (R-OBS-01)", pr))
|
||||
|
||||
for res in r_image_automation(s):
|
||||
docs.append((f"Flux {res['kind']} (R-UPD)", res))
|
||||
|
||||
return docs
|
||||
|
||||
|
||||
def emit(docs: list) -> str:
|
||||
"""Deterministic multi-document YAML text (golden-diffable)."""
|
||||
parts = []
|
||||
for comment, resource in docs:
|
||||
body = yaml.dump(
|
||||
resource,
|
||||
sort_keys=False,
|
||||
default_flow_style=False,
|
||||
width=4096,
|
||||
)
|
||||
parts.append(f"---\n# {comment}\n{body}".rstrip() + "\n")
|
||||
return "".join(parts)
|
||||
|
||||
|
||||
def load_sd(path: Path) -> dict:
|
||||
with path.open() as fh:
|
||||
return yaml.safe_load(fh)
|
||||
|
||||
|
||||
def render_file(path: Path) -> str:
|
||||
return emit(render(load_sd(path)))
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Self-test (deterministic, no cluster) — `render.py --selftest`
|
||||
# =============================================================================
|
||||
def _selftest() -> int:
|
||||
failures = []
|
||||
|
||||
def check(cond, msg):
|
||||
if not cond:
|
||||
failures.append(msg)
|
||||
|
||||
# 1) A minimal container SD renders the mandatory contract invariants.
|
||||
sd = {"name": "demo", "source": {"image": "nginx:1"}}
|
||||
docs = render(sd)
|
||||
kinds = [d[1]["kind"] for d in docs]
|
||||
check("Namespace" in kinds, "container: missing Namespace")
|
||||
check("NetworkPolicy" in kinds, "container: missing default-deny NetworkPolicy")
|
||||
check("Deployment" in kinds, "container: missing Deployment")
|
||||
|
||||
ns = next(d[1] for d in docs if d[1]["kind"] == "Namespace")
|
||||
check(
|
||||
ns["metadata"]["labels"].get(f"{LBL}/app-namespace") == "true",
|
||||
"namespace missing app-namespace=true (Kyverno hook)",
|
||||
)
|
||||
dep = next(d[1] for d in docs if d[1]["kind"] == "Deployment")
|
||||
plabels = dep["spec"]["template"]["metadata"]["labels"]
|
||||
check(
|
||||
plabels.get(f"{LBL}/egress-class") == "internet",
|
||||
"default egress-class must be internet (R-NET-05)",
|
||||
)
|
||||
check(
|
||||
plabels.get(f"{LBL}/auth-mode") in AUTH_MODES,
|
||||
"pod must carry a valid auth-mode label (R-IDP-02)",
|
||||
)
|
||||
ndp = next(d[1] for d in docs if d[1]["kind"] == "NetworkPolicy")
|
||||
check(
|
||||
ndp["spec"]["policyTypes"] == ["Ingress", "Egress"],
|
||||
"default-deny must cover both directions (R-NET-01)",
|
||||
)
|
||||
check(ndp["spec"]["ingress"] == [], "default-deny must deny all ingress (R-NET-01)")
|
||||
|
||||
# 2) egress: vpn flips ONLY the egress-class label (R-NET-07).
|
||||
internet_docs = render({"name": "x", "source": {"image": "a:1"}, "egress": "internet"})
|
||||
vpn_docs = render({"name": "x", "source": {"image": "a:1"}, "egress": "vpn"})
|
||||
i_pod = next(d[1] for d in internet_docs if d[1]["kind"] == "Deployment")
|
||||
v_pod = next(d[1] for d in vpn_docs if d[1]["kind"] == "Deployment")
|
||||
check(
|
||||
i_pod["spec"]["template"]["metadata"]["labels"][f"{LBL}/egress-class"] == "internet"
|
||||
and v_pod["spec"]["template"]["metadata"]["labels"][f"{LBL}/egress-class"] == "vpn",
|
||||
"egress class must render onto the pod label",
|
||||
)
|
||||
# Everything except the egress-class label must be identical (R-NET-07).
|
||||
i_txt = emit(internet_docs).replace("egress-class: internet", "EGRESS")
|
||||
v_txt = emit(vpn_docs).replace("egress-class: vpn", "EGRESS")
|
||||
check(i_txt == v_txt, "flipping egress class changed more than the egress path (R-NET-07)")
|
||||
|
||||
# 3) Determinism: same input twice -> byte-identical output.
|
||||
check(emit(render(sd)) == emit(render(sd)), "render is not deterministic")
|
||||
|
||||
# 4) dependencies -> one CiliumNetworkPolicy allow each (R-NET-02).
|
||||
dep_docs = render(
|
||||
{
|
||||
"name": "y",
|
||||
"source": {"image": "a:1"},
|
||||
"dependencies": [{"to": "db", "port": 5432}],
|
||||
}
|
||||
)
|
||||
cnps = [d[1] for d in dep_docs if d[1]["kind"] == "CiliumNetworkPolicy"]
|
||||
check(len(cnps) == 1, "one dependency must render one CiliumNetworkPolicy allow")
|
||||
|
||||
# 5) vm kind -> VirtualMachine + standalone DataVolume.
|
||||
vm_docs = render(
|
||||
{
|
||||
"name": "vmx",
|
||||
"kind": "vm",
|
||||
"source": {"vmTemplate": "quay.io/containerdisks/fedora:41"},
|
||||
"storage": [{"name": "rootdisk", "size": "6Gi", "class": "longhorn"}],
|
||||
}
|
||||
)
|
||||
vm_kinds = [d[1]["kind"] for d in vm_docs]
|
||||
check("VirtualMachine" in vm_kinds, "vm: missing VirtualMachine")
|
||||
check("DataVolume" in vm_kinds, "vm: missing CDI DataVolume")
|
||||
|
||||
if failures:
|
||||
for f in failures:
|
||||
sys.stderr.write(f"SELFTEST FAIL: {f}\n")
|
||||
return 1
|
||||
sys.stderr.write("render.py selftest: PASS\n")
|
||||
return 0
|
||||
|
||||
|
||||
def main(argv=None) -> int:
|
||||
ap = argparse.ArgumentParser(description="Render a ServiceDefinition to manifests.")
|
||||
ap.add_argument("input", nargs="?", help="ServiceDefinition YAML file")
|
||||
ap.add_argument("--all", metavar="DIR", help="Render every *.yaml ServiceDefinition in DIR")
|
||||
ap.add_argument("-o", "--out-dir", help="With --all: write <name>.yaml files into this dir")
|
||||
ap.add_argument("--selftest", action="store_true", help="Run the built-in self-test")
|
||||
args = ap.parse_args(argv)
|
||||
|
||||
if args.selftest:
|
||||
return _selftest()
|
||||
|
||||
if args.all:
|
||||
src = Path(args.all)
|
||||
files = sorted(p for p in src.glob("*.yaml") if p.name != "render.py")
|
||||
for f in files:
|
||||
text = render_file(f)
|
||||
if args.out_dir:
|
||||
outdir = Path(args.out_dir)
|
||||
outdir.mkdir(parents=True, exist_ok=True)
|
||||
name = load_sd(f)["name"]
|
||||
(outdir / f"{name}.yaml").write_text(text)
|
||||
else:
|
||||
sys.stdout.write(text)
|
||||
return 0
|
||||
|
||||
if not args.input:
|
||||
ap.error("provide a ServiceDefinition file, --all DIR, or --selftest")
|
||||
sys.stdout.write(render_file(Path(args.input)))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in new issue
Block a user