82b2024727
The in-cluster runner and the request file both load config/release-index.json, so a claimed workload's recommendation names the unit id and the declaration owner.
348 lines
17 KiB
Python
348 lines
17 KiB
Python
#!/usr/bin/env python3
|
|
"""CUST-WP-0071-T05 in-cluster weekly runner: read-only collection, join, retain.
|
|
|
|
Standard library only, so it runs on a pinned python:3.12-slim image with the
|
|
scripts mounted from a ConfigMap. It reads the Kubernetes API with the pod's
|
|
ServiceAccount (get/list only), queries Prometheus over plain HTTP, reuses the
|
|
existing owner code unchanged (railiance-cluster build_observation, custodian
|
|
reconcile join, weekly_allocation_report build_report/retain) and writes only
|
|
under the data directory. It never resizes, creates tasks or delivers messages.
|
|
|
|
Week identity is the ISO week of (run time - 6h); the telemetry window always
|
|
ends at Monday 06:00 UTC of that week, so consecutive reports are exactly
|
|
604800 s apart regardless of DST or scheduler jitter. 06:00 UTC is at or before
|
|
the Monday 08:00 Europe/Berlin schedule in both CET and CEST and keeps the
|
|
window start as close as possible to the seven-day Prometheus retention edge. A week whose report is already
|
|
retained is a no-op success (idempotent retry); a failed attempt leaves no
|
|
report, so the Job's retry or the next manual run collects again.
|
|
|
|
Delivery (founder decision 2026-10-21, "Hub message + repo tasks"): after a
|
|
week is retained, one State Hub message goes from custodian-allocation-review
|
|
to the-custodian with a short summary, report path and sha256. The message id
|
|
is recorded in receipts.jsonl; a week that already has a message id is never
|
|
posted again, and a retained-but-undelivered week is posted by the next run.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
from datetime import datetime, timedelta, timezone
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import ssl
|
|
import sys
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
HERE = Path(__file__).resolve().parent
|
|
SA_DIR = Path("/var/run/secrets/kubernetes.io/serviceaccount")
|
|
WEEK = re.compile(r"^(\d{4})-W(\d{2})\.json$")
|
|
# Exactly the reads the collector performs; mirrored by the ClusterRole.
|
|
API_PATHS = {
|
|
"nodes": "/api/v1/nodes",
|
|
"pods": "/api/v1/pods",
|
|
"pvcs": "/api/v1/persistentvolumeclaims",
|
|
"pvs": "/api/v1/persistentvolumes",
|
|
"storageclasses": "/apis/storage.k8s.io/v1/storageclasses",
|
|
}
|
|
METRICS_PATHS = {
|
|
"node_top": "/apis/metrics.k8s.io/v1beta1/nodes",
|
|
"pod_top": "/apis/metrics.k8s.io/v1beta1/pods",
|
|
}
|
|
|
|
|
|
def load(path: Path, name: str):
|
|
spec = importlib.util.spec_from_file_location(name, path)
|
|
if spec is None or spec.loader is None:
|
|
raise RuntimeError(f"cannot load {path}")
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def sha256(path: Path) -> str:
|
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
|
|
|
|
|
def iso(value: datetime) -> str:
|
|
return value.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
|
|
|
|
|
|
WINDOW_END_HOUR = 6
|
|
|
|
|
|
def week_of(now: datetime) -> tuple[str, datetime]:
|
|
"""ISO week label and its Monday 06:00 UTC (the telemetry window end)."""
|
|
shifted = now.astimezone(timezone.utc) - timedelta(hours=WINDOW_END_HOUR)
|
|
year, week, weekday = shifted.isocalendar()
|
|
monday = (shifted - timedelta(days=weekday - 1)).replace(hour=0, minute=0, second=0, microsecond=0)
|
|
return f"{year}-W{week:02d}", monday + timedelta(hours=WINDOW_END_HOUR)
|
|
|
|
|
|
def missed_weeks(previous: str | None, current: str) -> list[str]:
|
|
if previous is None:
|
|
return []
|
|
def monday(label):
|
|
year, week = int(label[:4]), int(label[6:8])
|
|
return datetime.fromisocalendar(year, week, 1).replace(tzinfo=timezone.utc, hour=WINDOW_END_HOUR)
|
|
out, cursor = [], monday(previous) + timedelta(days=7)
|
|
while cursor < monday(current):
|
|
out.append(week_of(cursor)[0])
|
|
cursor += timedelta(days=7)
|
|
return out
|
|
|
|
|
|
def previous_report(reports: Path, label: str) -> Path | None:
|
|
earlier = sorted(p for p in reports.glob("*.json") if WEEK.match(p.name) and p.stem < label)
|
|
return earlier[-1] if earlier else None
|
|
|
|
|
|
class Kube:
|
|
"""Minimal in-cluster API reader; the token is never logged or persisted."""
|
|
|
|
def __init__(self, base: str, token_path: Path, ca_path: Path):
|
|
self.base = base.rstrip("/")
|
|
self.token_path = token_path
|
|
self.context = ssl.create_default_context(cafile=str(ca_path))
|
|
|
|
def get(self, path: str):
|
|
request = urllib.request.Request(self.base + path, headers={
|
|
"Authorization": "Bearer " + self.token_path.read_text().strip(),
|
|
"Accept": "application/json"})
|
|
with urllib.request.urlopen(request, context=self.context, timeout=60) as response:
|
|
return json.load(response)
|
|
|
|
|
|
def top_lines(metrics: dict, *, pods: bool, observer) -> str:
|
|
"""Render metrics.k8s.io lists in the `kubectl top` shape parse_top reads."""
|
|
lines = []
|
|
for item in metrics.get("items") or []:
|
|
meta = item.get("metadata") or {}
|
|
if pods:
|
|
cpu = sum(observer.cpu_millicores((c.get("usage") or {}).get("cpu")) for c in item.get("containers") or [])
|
|
mem = sum(observer.bytes_value((c.get("usage") or {}).get("memory")) for c in item.get("containers") or [])
|
|
lines.append(f"{meta.get('namespace')} {meta.get('name')} {cpu}m {mem}")
|
|
else:
|
|
usage = item.get("usage") or {}
|
|
cpu = observer.cpu_millicores(usage.get("cpu"))
|
|
mem = observer.bytes_value(usage.get("memory"))
|
|
lines.append(f"{meta.get('name')} {cpu}m 0% {mem} 0%")
|
|
return "\n".join(lines) + ("\n" if lines else "")
|
|
|
|
|
|
def collect_snapshots(get, observer) -> dict:
|
|
snapshots = {name: get(path) for name, path in API_PATHS.items()}
|
|
for name, path in METRICS_PATHS.items():
|
|
try:
|
|
snapshots[name] = top_lines(get(path), pods=name == "pod_top", observer=observer)
|
|
except (OSError, ValueError, urllib.error.URLError):
|
|
snapshots[name] = None # recorded by build_observation as a measurement gap
|
|
return snapshots
|
|
|
|
|
|
def prometheus_query(base: str, promql: str, at: datetime, opener=urllib.request.urlopen) -> dict:
|
|
"""Return Prometheus' JSON response, including error bodies, never raise on 4xx/5xx."""
|
|
url = base.rstrip("/") + "/api/v1/query?" + urllib.parse.urlencode(
|
|
{"query": promql, "time": f"{at.timestamp():.3f}"})
|
|
try:
|
|
with opener(url, timeout=120) as response:
|
|
return json.load(response)
|
|
except urllib.error.HTTPError as exc:
|
|
try:
|
|
return json.loads(exc.read())
|
|
except ValueError:
|
|
return {"status": "error", "errorType": "http", "error": f"HTTP {exc.code}"}
|
|
except (OSError, ValueError) as exc:
|
|
return {"status": "error", "errorType": "transport", "error": type(exc).__name__}
|
|
|
|
|
|
def write_json(path: Path, value) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
temporary = path.with_name(path.name + ".tmp")
|
|
temporary.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n")
|
|
os.replace(temporary, path)
|
|
|
|
|
|
def append_receipt(data: Path, receipt: dict) -> None:
|
|
data.mkdir(parents=True, exist_ok=True)
|
|
with (data / "receipts.jsonl").open("a") as stream:
|
|
stream.write(json.dumps(receipt, sort_keys=True) + "\n")
|
|
stream.flush()
|
|
os.fsync(stream.fileno())
|
|
print(json.dumps(receipt, sort_keys=True))
|
|
|
|
|
|
def delivered_message(data: Path, week: str) -> str | None:
|
|
path = data / "receipts.jsonl"
|
|
if not path.exists():
|
|
return None
|
|
for line in reversed(path.read_text().splitlines()):
|
|
try:
|
|
entry = json.loads(line)
|
|
except ValueError:
|
|
continue
|
|
if entry.get("week") == week and entry.get("message_id"):
|
|
return entry["message_id"]
|
|
return None
|
|
|
|
|
|
def summary(receipt: dict) -> dict:
|
|
report = json.loads(Path(receipt["report"]).read_text())
|
|
decisions: dict[str, int] = {}
|
|
for item in report["recommendations"]:
|
|
decisions[item["decision"]] = decisions.get(item["decision"], 0) + 1
|
|
return {"week": receipt["week"], "report": receipt["report"], "report_sha256": receipt["report_sha256"],
|
|
"capture_end": report["period"]["capture_end"], "workloads": len(report["workloads"]),
|
|
"namespaces_with_metrics": len(report["namespace_context"]),
|
|
"recommendations": dict(sorted(decisions.items())),
|
|
"comparison_compatible": report["comparison"]["compatible"],
|
|
"measurement_gaps": len(report["measurement_gaps"]),
|
|
"failed_queries": sorted(n for n, q in report["queries"].items() if q.get("response_status") != "success"),
|
|
"next_step": "Custodian session reviews and records owner-repo tasks (ADR-001); no automatic resizing."}
|
|
|
|
|
|
def hub_post(hub: str, body: dict) -> str:
|
|
request = urllib.request.Request(hub.rstrip("/") + "/messages/", data=json.dumps(body).encode(),
|
|
headers={"Content-Type": "application/json"}, method="POST")
|
|
with urllib.request.urlopen(request, timeout=30) as response:
|
|
return str(json.load(response)["id"])
|
|
|
|
|
|
def deliver(data: Path, receipt: dict, hub: str | None, post=hub_post) -> tuple[dict, bool]:
|
|
"""Post one hub message per retained week; returns (receipt, ok)."""
|
|
if receipt.get("status") not in {"retained", "already-retained"}:
|
|
return receipt, True
|
|
existing = delivered_message(data, receipt["week"])
|
|
if existing:
|
|
return {**receipt, "delivery": "already-posted", "message_id": existing}, True
|
|
if not hub:
|
|
return {**receipt, "delivery": "not configured"}, True
|
|
content = summary(receipt)
|
|
body = {"from_agent": "custodian-allocation-review", "to_agent": "the-custodian",
|
|
"subject": f"Weekly allocation assessment {receipt['week']}",
|
|
"body": json.dumps(content, sort_keys=True)}
|
|
try:
|
|
return {**receipt, "delivery": "posted", "message_id": post(hub, body)}, True
|
|
except (OSError, ValueError, KeyError) as exc:
|
|
return {**receipt, "delivery": f"failed: {type(exc).__name__}"}, False
|
|
|
|
|
|
def run(args, *, now: datetime, get, query) -> dict:
|
|
sources = {
|
|
"weekly_allocation_report": args.source_dir / "weekly_allocation_report.py",
|
|
"reconcile_railiance_allocation": args.source_dir / "reconcile_railiance_allocation.py",
|
|
"observe_cluster_resources": args.observer,
|
|
"weekly_allocation_incluster": Path(__file__).resolve(),
|
|
"queries": args.queries, "ownership": args.ownership, "inventory": args.inventory,
|
|
}
|
|
release_index_path = getattr(args, "release_index", None)
|
|
if release_index_path is None:
|
|
raise ValueError("release index must be bound")
|
|
release_index_path = Path(release_index_path)
|
|
sources["release_index"] = release_index_path
|
|
release_index = json.loads(release_index_path.read_text())
|
|
if not isinstance(release_index, dict) or (
|
|
"units" not in release_index and "findings" not in release_index
|
|
):
|
|
raise ValueError("release index is not a composed index or coverage report")
|
|
weekly = load(sources["weekly_allocation_report"], "weekly_allocation_report")
|
|
reconcile = load(sources["reconcile_railiance_allocation"], "reconcile_railiance_allocation")
|
|
observer = load(sources["observe_cluster_resources"], "observe_cluster_resources")
|
|
label, window_end = week_of(now)
|
|
reports = args.data_dir / "reports"
|
|
target = reports / f"{label}.json"
|
|
receipt = {"schema": "custodian.weekly-allocation-receipt.v1", "week": label,
|
|
"run_at": iso(now), "report": str(target), "delivery": "not emitted",
|
|
"acknowledgment": None}
|
|
if target.exists():
|
|
return {**receipt, "status": "already-retained", "report_sha256": sha256(target)}
|
|
previous_path = previous_report(reports, label)
|
|
previous_label = previous_path.stem if previous_path else None
|
|
|
|
observation = observer.build_observation(collect_snapshots(get, observer), iso(now))
|
|
owners = json.loads(args.ownership.read_text())
|
|
allocation = reconcile.join(observation, owners)
|
|
spec = json.loads(args.queries.read_text())
|
|
if spec.get("schema") != "custodian.weekly-allocation-queries.v1":
|
|
raise ValueError("unsupported query specification")
|
|
telemetry = {"captured_at": window_end.isoformat(), "window": spec["window"], "step": spec["step"],
|
|
"source": args.prometheus_url,
|
|
"queries": {name: {"promql": promql, "response": query(promql, window_end)}
|
|
for name, promql in sorted(spec["queries"].items())}}
|
|
provenance = {
|
|
"schema": "custodian.allocation-capture-provenance.v1", "workplan": "CUST-WP-0071-T05",
|
|
"source_observation": observation["captured_at"], "read_only": True,
|
|
"secret_surfaces_read": False, "executor": args.executor,
|
|
"kubernetes_reads": sorted(API_PATHS.values()) + sorted(METRICS_PATHS.values()),
|
|
"prometheus": args.prometheus_url, "telemetry_window_end": window_end.isoformat(),
|
|
"inputs": [{"name": n, "path": str(p), "sha256": sha256(p)} for n, p in sorted(sources.items())],
|
|
}
|
|
inventory = json.loads(args.inventory.read_text())
|
|
inputs = {"cluster": observation, "allocation": allocation, "telemetry": telemetry,
|
|
"inventory": inventory, "provenance": provenance}
|
|
attempt = args.data_dir / "runs" / label / iso(now).replace(":", "")
|
|
for name, value in inputs.items():
|
|
write_json(attempt / f"{name}.json", value)
|
|
report = weekly.build_report(
|
|
**inputs, as_of=iso(now), max_age_hours=args.max_age_hours,
|
|
previous=json.loads(previous_path.read_text()) if previous_path else None,
|
|
reviews=json.loads(args.reviews.read_text()) if args.reviews and args.reviews.exists() else None,
|
|
release_index=release_index)
|
|
weekly.retain(target, report)
|
|
failed = sorted(n for n, q in telemetry["queries"].items() if q["response"].get("status") != "success")
|
|
return {**receipt, "status": "retained", "report_sha256": sha256(target),
|
|
"inputs_dir": str(attempt), "previous_week": previous_label,
|
|
"missed_weeks": missed_weeks(previous_label, label),
|
|
"comparison_compatible": report["comparison"]["compatible"],
|
|
"workloads": len(report["workloads"]), "failed_queries": failed,
|
|
"measurement_gaps": len(report["measurement_gaps"])}
|
|
|
|
|
|
def parser() -> argparse.ArgumentParser:
|
|
env = os.environ.get
|
|
p = argparse.ArgumentParser(description=__doc__)
|
|
p.add_argument("--data-dir", type=Path, default=Path(env("ASSESSMENT_DATA_DIR", "/data")))
|
|
p.add_argument("--source-dir", type=Path, default=Path(env("ASSESSMENT_SOURCE_DIR", str(HERE))))
|
|
p.add_argument("--observer", type=Path, default=Path(env("ASSESSMENT_OBSERVER",
|
|
str(Path(env("ASSESSMENT_SOURCE_DIR", str(HERE))) / "observe_cluster_resources.py"))))
|
|
p.add_argument("--queries", type=Path, default=Path(env("ASSESSMENT_QUERIES", str(HERE.parent / "config/weekly-allocation-queries.json"))))
|
|
p.add_argument("--ownership", type=Path, default=env("ASSESSMENT_OWNERSHIP") and Path(env("ASSESSMENT_OWNERSHIP")))
|
|
p.add_argument("--inventory", type=Path, default=env("ASSESSMENT_INVENTORY") and Path(env("ASSESSMENT_INVENTORY")))
|
|
p.add_argument("--reviews", type=Path, default=env("ASSESSMENT_REVIEWS") and Path(env("ASSESSMENT_REVIEWS")))
|
|
p.add_argument("--release-index", type=Path, default=Path(env(
|
|
"ASSESSMENT_RELEASE_INDEX", str(HERE.parent / "config/release-index.json"))))
|
|
p.add_argument("--prometheus-url", default=env("PROMETHEUS_URL", "http://telemetry-prometheus.telemetry.svc:9090"))
|
|
p.add_argument("--hub-url", default=env("STATE_HUB_URL"), help="State Hub API; unset disables delivery")
|
|
p.add_argument("--executor", default=env("ASSESSMENT_EXECUTOR", "unspecified"))
|
|
p.add_argument("--max-age-hours", type=float, default=float(env("ASSESSMENT_MAX_AGE_HOURS", "48")))
|
|
return p
|
|
|
|
|
|
def main() -> int:
|
|
args = parser().parse_args()
|
|
now = datetime.now(timezone.utc)
|
|
label = week_of(now)[0]
|
|
try:
|
|
if not args.ownership or not args.inventory:
|
|
raise ValueError("ownership and inventory inputs must be bound")
|
|
host = os.environ["KUBERNETES_SERVICE_HOST"]
|
|
port = os.environ.get("KUBERNETES_SERVICE_PORT", "443")
|
|
kube = Kube(f"https://{host}:{port}", SA_DIR / "token", SA_DIR / "ca.crt")
|
|
receipt = run(args, now=now, get=kube.get,
|
|
query=lambda promql, at: prometheus_query(args.prometheus_url, promql, at))
|
|
except Exception as exc: # noqa: BLE001 - every failure must leave a receipt
|
|
append_receipt(args.data_dir, {"schema": "custodian.weekly-allocation-receipt.v1",
|
|
"week": label, "run_at": iso(now), "status": "failed",
|
|
"error": f"{type(exc).__name__}: {exc}"[:500]})
|
|
return 1
|
|
receipt, delivered = deliver(args.data_dir, receipt, args.hub_url)
|
|
append_receipt(args.data_dir, receipt)
|
|
return 0 if delivered else 1 # undelivered: Job retry posts it (report already retained)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|