216 lines
7.5 KiB
Python
216 lines
7.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Render and revalidate the two-workload Hermes WebUI Flux handoff."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import difflib
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
DEFAULT_IMAGE = "registry.bstein.dev/bstein/hermes-webui"
|
|
DIGEST_PATTERN = re.compile(r"^sha256:[0-9a-f]{64}$")
|
|
REVISION_PATTERN = re.compile(r"^[0-9a-f]{40}$")
|
|
BUILD_PATTERN = re.compile(r"^[1-9][0-9]*$")
|
|
DESTINATION_PATTERN = re.compile(
|
|
r"^registry\.bstein\.dev/bstein/hermes-webui:"
|
|
r"git-([0-9a-f]{40})-build-([1-9][0-9]*)$"
|
|
)
|
|
|
|
|
|
def validated(value: str, pattern: re.Pattern[str], label: str) -> str:
|
|
"""Return a normalized value when it matches the release contract."""
|
|
normalized = value.strip()
|
|
if not pattern.fullmatch(normalized):
|
|
raise ValueError(f"invalid {label}: expected {pattern.pattern}")
|
|
return normalized
|
|
|
|
|
|
def validate_destination(
|
|
destination: str, source_revision: str, build_number: str
|
|
) -> tuple[str, str]:
|
|
"""Bind one unique build tag to the reviewed revision and Jenkins build."""
|
|
revision = validated(source_revision, REVISION_PATTERN, "source revision")
|
|
build = validated(build_number, BUILD_PATTERN, "build number")
|
|
match = DESTINATION_PATTERN.fullmatch(destination.strip())
|
|
if not match or match.groups() != (revision, build):
|
|
raise ValueError(
|
|
"destination must bind the reviewed revision and unique Jenkins build"
|
|
)
|
|
return revision, build
|
|
|
|
|
|
def render_workload(
|
|
source: str,
|
|
digest: str,
|
|
*,
|
|
kind: str,
|
|
name: str,
|
|
image: str = DEFAULT_IMAGE,
|
|
) -> str:
|
|
"""Replace one WebUI image in one exact Flux workload without reformatting."""
|
|
digest = validated(digest, DIGEST_PATTERN, "image digest")
|
|
identity = re.compile(
|
|
rf"\A(?:#[^\n]*\n)*apiVersion: apps/v1\nkind: {re.escape(kind)}\n"
|
|
rf"metadata:\n name: {re.escape(name)}\n"
|
|
)
|
|
if not identity.search(source):
|
|
raise ValueError(f"Flux target identity changed: expected {kind}/{name}")
|
|
lines = source.splitlines(keepends=True)
|
|
matches: list[int] = []
|
|
suffixes: dict[int, str] = {}
|
|
for index, line in enumerate(lines):
|
|
stripped = line.strip()
|
|
if not stripped.startswith("image: "):
|
|
continue
|
|
value, separator, comment = stripped.removeprefix("image: ").partition(" #")
|
|
current_image, at, current_digest = value.rpartition("@")
|
|
if not at or not re.fullmatch(
|
|
rf"{re.escape(image)}(?::[A-Za-z0-9_][A-Za-z0-9_.-]{{0,127}})?",
|
|
current_image,
|
|
):
|
|
continue
|
|
validated(current_digest, DIGEST_PATTERN, "current Flux image digest")
|
|
matches.append(index)
|
|
suffixes[index] = f" #{comment}" if separator else ""
|
|
if len(matches) != 1:
|
|
raise ValueError(
|
|
f"expected exactly one {image!r} image in {kind}/{name}; "
|
|
f"found {len(matches)}"
|
|
)
|
|
index = matches[0]
|
|
newline = "\n" if lines[index].endswith("\n") else ""
|
|
prefix = lines[index][: len(lines[index]) - len(lines[index].lstrip())]
|
|
lines[index] = f"{prefix}image: {image}@{digest}{suffixes[index]}{newline}"
|
|
return "".join(lines)
|
|
|
|
|
|
def _targets(chat_manifest: Path, dashboard_manifest: Path):
|
|
return (
|
|
(
|
|
chat_manifest,
|
|
"StatefulSet",
|
|
"hermes-chat-tenant",
|
|
"hermes-chat-statefulset.yaml",
|
|
),
|
|
(
|
|
dashboard_manifest,
|
|
"Deployment",
|
|
"hermes",
|
|
"hermes-dashboard-deployment.yaml",
|
|
),
|
|
)
|
|
|
|
|
|
def _metadata(
|
|
digest: str, source_revision: str, build_number: str, destination: str
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"build_number": build_number,
|
|
"digest": digest,
|
|
"flux_image": f"{DEFAULT_IMAGE}@{digest}",
|
|
"flux_targets": [
|
|
"apps/StatefulSet/hermes/hermes-chat-tenant",
|
|
"apps/Deployment/hermes/hermes",
|
|
],
|
|
"image": DEFAULT_IMAGE,
|
|
"published_tag": destination,
|
|
"source_revision": source_revision,
|
|
}
|
|
|
|
|
|
def _rendered_and_patch(
|
|
digest: str, chat_manifest: Path, dashboard_manifest: Path
|
|
) -> tuple[dict[str, str], str]:
|
|
rendered_targets: dict[str, str] = {}
|
|
patch_parts: list[str] = []
|
|
for path, kind, name, artifact_name in _targets(chat_manifest, dashboard_manifest):
|
|
source = path.read_text(encoding="utf-8")
|
|
rendered = render_workload(source, digest, kind=kind, name=name)
|
|
rendered_targets[artifact_name] = rendered
|
|
patch_parts.append(
|
|
"".join(
|
|
difflib.unified_diff(
|
|
source.splitlines(keepends=True),
|
|
rendered.splitlines(keepends=True),
|
|
fromfile=f"a/services/hermes/{path.name}",
|
|
tofile=f"b/services/hermes/{path.name}",
|
|
)
|
|
)
|
|
)
|
|
return rendered_targets, "".join(patch_parts)
|
|
|
|
|
|
def write_release_artifacts(
|
|
*,
|
|
digest: str,
|
|
source_revision: str,
|
|
build_number: str,
|
|
destination: str,
|
|
chat_manifest: Path,
|
|
dashboard_manifest: Path,
|
|
output_dir: Path,
|
|
) -> dict[str, Any]:
|
|
"""Write two rendered Flux targets, one patch, and credential-free evidence."""
|
|
digest = validated(digest, DIGEST_PATTERN, "image digest")
|
|
source_revision, build_number = validate_destination(
|
|
destination, source_revision, build_number
|
|
)
|
|
rendered_targets, patch = _rendered_and_patch(
|
|
digest, chat_manifest, dashboard_manifest
|
|
)
|
|
if not patch:
|
|
raise ValueError("published digest already matches every Flux target")
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
for artifact_name, rendered in rendered_targets.items():
|
|
(output_dir / artifact_name).write_text(rendered, encoding="utf-8")
|
|
(output_dir / "hermes-webui-image-update.patch").write_text(patch, encoding="utf-8")
|
|
metadata = _metadata(digest, source_revision, build_number, destination)
|
|
(output_dir / "hermes-webui-image.json").write_text(
|
|
json.dumps(metadata, indent=2, sort_keys=True) + "\n", encoding="utf-8"
|
|
)
|
|
return metadata
|
|
|
|
|
|
def validate_release_artifacts(
|
|
*,
|
|
digest: str,
|
|
source_revision: str,
|
|
build_number: str,
|
|
destination: str,
|
|
chat_manifest: Path,
|
|
dashboard_manifest: Path,
|
|
output_dir: Path,
|
|
) -> None:
|
|
"""Revalidate the exact successful-build evidence without rewriting it."""
|
|
digest = validated(digest, DIGEST_PATTERN, "image digest")
|
|
source_revision, build_number = validate_destination(
|
|
destination, source_revision, build_number
|
|
)
|
|
expected_names = {
|
|
"hermes-webui-image.json",
|
|
"hermes-webui-image-update.patch",
|
|
"hermes-chat-statefulset.yaml",
|
|
"hermes-dashboard-deployment.yaml",
|
|
}
|
|
entries = list(output_dir.iterdir())
|
|
if {entry.name for entry in entries} != expected_names or not all(
|
|
entry.is_file() and not entry.is_symlink() for entry in entries
|
|
):
|
|
raise ValueError("release output must contain exactly four evidence files")
|
|
rendered_targets, patch = _rendered_and_patch(
|
|
digest, chat_manifest, dashboard_manifest
|
|
)
|
|
metadata = _metadata(digest, source_revision, build_number, destination)
|
|
expected = {
|
|
"hermes-webui-image.json": json.dumps(metadata, indent=2, sort_keys=True)
|
|
+ "\n",
|
|
"hermes-webui-image-update.patch": patch,
|
|
**rendered_targets,
|
|
}
|
|
for name, expected_text in expected.items():
|
|
if (output_dir / name).read_text(encoding="utf-8") != expected_text:
|
|
raise ValueError(f"release evidence is incomplete or mismatched: {name}")
|