b084150906
Serving-core and Kourier each carry the Deployments that exist in that upstream file. The render test expects the label and still rejects any other drift. RELREG-WP-0002-T01 Assistant: grok Assistant-Session: 01a15f9f-3d77-7d21-911b-d93c86f9d14f
185 lines
8.2 KiB
Python
185 lines
8.2 KiB
Python
"""Offline proof that install/knative renders the declared CPU requests and HPA caps.
|
|
|
|
Runs install/knative/render.sh (download + SHA-256 verification + kustomize;
|
|
no cluster contact) and compares the result with the upstream assets: the six
|
|
declared CPU requests and HPA maxReplicas change, each of those Deployments
|
|
carries railiance.io/release-unit=knative, the Kourier Service is ClusterIP,
|
|
the gateway image is the pinned Envoy digest, and nothing else differs.
|
|
Skips when the upstream assets cannot be fetched or kubectl is missing.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import hashlib
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
try:
|
|
import yaml
|
|
except ImportError: # pragma: no cover
|
|
yaml = None
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
KNATIVE = ROOT / "install" / "knative"
|
|
RAIL_KNATIVE = Path(os.environ.get("RAIL_KNATIVE_DIR", ROOT.parent / "rail-knative"))
|
|
DECLARED = { # (namespace, deployment, container) -> cpu request
|
|
("knative-serving", "activator", "activator"): "10m",
|
|
("knative-serving", "autoscaler", "autoscaler"): "10m",
|
|
("knative-serving", "controller", "controller"): "10m",
|
|
("knative-serving", "webhook", "webhook"): "10m",
|
|
("knative-serving", "net-kourier-controller", "controller"): "10m",
|
|
("kourier-system", "3scale-kourier-gateway", "kourier-gateway"): "10m",
|
|
}
|
|
DECLARED_HPA = { # (namespace, hpa) -> maxReplicas
|
|
("knative-serving", "activator"): 3,
|
|
("knative-serving", "webhook"): 2,
|
|
("kourier-system", "3scale-kourier-gateway"): 3,
|
|
}
|
|
DECLARED_RELEASE_UNIT = { # (namespace, deployment) -> label
|
|
("knative-serving", "activator"): "knative",
|
|
("knative-serving", "autoscaler"): "knative",
|
|
("knative-serving", "controller"): "knative",
|
|
("knative-serving", "webhook"): "knative",
|
|
("knative-serving", "net-kourier-controller"): "knative",
|
|
("kourier-system", "3scale-kourier-gateway"): "knative",
|
|
}
|
|
|
|
|
|
def lock() -> dict[str, str]:
|
|
lines = (KNATIVE / "release-lock.env").read_text().splitlines()
|
|
return dict(line.split("=", 1) for line in lines if "=" in line)
|
|
|
|
|
|
def load(path: Path) -> list[dict]:
|
|
return [d for d in yaml.safe_load_all(path.read_text()) if d]
|
|
|
|
|
|
def key(doc: dict) -> tuple:
|
|
meta = doc["metadata"]
|
|
return (doc["apiVersion"], doc["kind"], meta.get("namespace"), meta["name"])
|
|
|
|
|
|
def patch_requests(path: Path) -> dict[tuple, str]:
|
|
out = {}
|
|
for doc in load(path):
|
|
for c in doc["spec"]["template"]["spec"]["containers"]:
|
|
out[(doc["metadata"]["namespace"], doc["metadata"]["name"], c["name"])] = c["resources"]["requests"]["cpu"]
|
|
return out
|
|
|
|
|
|
def patch_hpa_caps(path: Path) -> dict[tuple, int]:
|
|
return {(d["metadata"]["namespace"], d["metadata"]["name"]): d["spec"]["maxReplicas"] for d in load(path)}
|
|
|
|
|
|
def patch_release_unit(path: Path) -> dict[tuple, str]:
|
|
return {
|
|
(d["metadata"]["namespace"], d["metadata"]["name"]): d["metadata"]["labels"]["railiance.io/release-unit"]
|
|
for d in load(path)
|
|
}
|
|
|
|
|
|
@unittest.skipIf(yaml is None, "PyYAML not installed")
|
|
class DeclarationTests(unittest.TestCase):
|
|
def test_overlays_carry_the_declared_requests(self) -> None:
|
|
got = {}
|
|
for overlay in ("serving-core", "kourier"):
|
|
got.update(patch_requests(KNATIVE / "overlays" / overlay / "cpu-requests.patch.yaml"))
|
|
self.assertEqual(DECLARED, got)
|
|
|
|
def test_overlays_carry_the_declared_hpa_caps(self) -> None:
|
|
got = {}
|
|
for overlay in ("serving-core", "kourier"):
|
|
got.update(patch_hpa_caps(KNATIVE / "overlays" / overlay / "hpa-caps.patch.yaml"))
|
|
self.assertEqual(DECLARED_HPA, got)
|
|
|
|
def test_verify_checks_the_declared_requests(self) -> None:
|
|
rows = re.search(r"<<'LIST'\n(.*?)\nLIST", (KNATIVE / "verify.sh").read_text(), re.S).group(1)
|
|
got = {tuple(r.split()[:3]): r.split()[3] for r in rows.splitlines()}
|
|
self.assertEqual(DECLARED, got)
|
|
|
|
def test_verify_checks_the_declared_hpa_caps(self) -> None:
|
|
rows = re.search(r"<<'HPAS'\n(.*?)\nHPAS", (KNATIVE / "verify.sh").read_text(), re.S).group(1)
|
|
got = {tuple(r.split()[:2]): int(r.split()[2]) for r in rows.splitlines()}
|
|
self.assertEqual(DECLARED_HPA, got)
|
|
|
|
def test_matches_rail_knative_declaration(self) -> None:
|
|
source = RAIL_KNATIVE / "substrate" / "v1.22.0" / "cpu-requests.patch.yaml"
|
|
if not source.exists():
|
|
self.skipTest(f"rail-knative not checked out at {RAIL_KNATIVE}")
|
|
self.assertEqual(DECLARED, patch_requests(source))
|
|
self.assertEqual(DECLARED_HPA, patch_hpa_caps(source.with_name("hpa-caps.patch.yaml")))
|
|
|
|
def test_release_unit_label_matches_rail_knative(self) -> None:
|
|
source = RAIL_KNATIVE / "substrate" / "v1.22.0" / "release-unit.patch.yaml"
|
|
if not source.exists():
|
|
self.skipTest(f"rail-knative not checked out at {RAIL_KNATIVE}")
|
|
got = {}
|
|
for overlay in ("serving-core", "kourier"):
|
|
got.update(patch_release_unit(KNATIVE / "overlays" / overlay / "release-unit.patch.yaml"))
|
|
self.assertEqual(DECLARED_RELEASE_UNIT, got)
|
|
self.assertEqual(DECLARED_RELEASE_UNIT, patch_release_unit(source))
|
|
|
|
|
|
@unittest.skipIf(yaml is None or shutil.which("kubectl") is None, "needs PyYAML and kubectl")
|
|
class RenderTests(unittest.TestCase):
|
|
stage: Path
|
|
|
|
@classmethod
|
|
def setUpClass(cls) -> None:
|
|
cls.stage = Path(os.environ.get("KNATIVE_STAGE_DIR") or tempfile.mkdtemp(prefix="knative-render-"))
|
|
result = subprocess.run([str(KNATIVE / "render.sh"), str(cls.stage)], capture_output=True, text=True)
|
|
if result.returncode != 0:
|
|
if "checksum mismatch" in result.stderr:
|
|
raise AssertionError(result.stderr)
|
|
raise unittest.SkipTest(f"render.sh failed (offline?): {result.stderr.strip()[:200]}")
|
|
|
|
def compare(self, upstream_file: str, rendered_file: str) -> dict[tuple, dict]:
|
|
upstream = {key(d): d for d in load(self.stage / upstream_file)}
|
|
rendered = {key(d): d for d in load(self.stage / rendered_file)}
|
|
self.assertEqual(set(upstream), set(rendered))
|
|
return {k: (upstream[k], rendered[k]) for k in upstream}
|
|
|
|
def check(self, upstream_file: str, rendered_file: str) -> set:
|
|
seen = set()
|
|
envoy = lock()["ENVOY_IMAGE"]
|
|
for k, (up, out) in self.compare(upstream_file, rendered_file).items():
|
|
expected = copy.deepcopy(up)
|
|
if k[1] == "Deployment":
|
|
if (k[2], k[3]) in DECLARED_RELEASE_UNIT:
|
|
labels = expected.setdefault("metadata", {}).get("labels") or {}
|
|
labels["railiance.io/release-unit"] = DECLARED_RELEASE_UNIT[(k[2], k[3])]
|
|
expected["metadata"]["labels"] = labels
|
|
for c in expected["spec"]["template"]["spec"]["containers"]:
|
|
ident = (k[2], k[3], c["name"])
|
|
if ident in DECLARED:
|
|
c["resources"]["requests"]["cpu"] = DECLARED[ident]
|
|
seen.add(ident)
|
|
if c["image"].startswith("docker.io/envoyproxy/envoy:"):
|
|
c["image"] = envoy
|
|
if k[1] == "HorizontalPodAutoscaler" and k[2:] in DECLARED_HPA:
|
|
expected["spec"]["maxReplicas"] = DECLARED_HPA[k[2:]]
|
|
seen.add(k[2:])
|
|
if k[1:] == ("Service", "kourier-system", "kourier"):
|
|
expected["spec"]["type"] = "ClusterIP"
|
|
self.assertEqual(expected, out, f"{k} differs beyond the declared changes")
|
|
return seen
|
|
|
|
def test_crds_are_the_pinned_upstream_file(self) -> None:
|
|
digest = hashlib.sha256((self.stage / "crds.yaml").read_bytes()).hexdigest()
|
|
self.assertEqual(lock()["SERVING_CRDS_SHA256"], digest)
|
|
for doc in load(self.stage / "crds.yaml"):
|
|
self.assertEqual("CustomResourceDefinition", doc["kind"])
|
|
|
|
def test_rendered_carries_declared_requests_and_nothing_else_changes(self) -> None:
|
|
seen = self.check("core.yaml", "serving-core.rendered.yaml")
|
|
seen |= self.check("kourier.yaml", "kourier.rendered.yaml")
|
|
self.assertEqual(set(DECLARED) | set(DECLARED_HPA), seen)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|