9edd3151ac
The in-cluster runner posts one State Hub message (custodian-allocation-review -> the-custodian) per newly retained week, records the message id in receipts.jsonl, never reposts, and retries an undelivered week. New weekly_allocation_check.py flags a failed latest receipt or a missing report for the due ISO week (Monday 12:00 Europe/Berlin); security_flow_daily.py notifies once per distinct flag set and stays inactive until a receipts source is configured. Founder decision 2026-10-21 (T05 follow-up answers). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
217 lines
12 KiB
Python
217 lines
12 KiB
Python
import argparse
|
|
from datetime import datetime, timezone
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
OBSERVER = Path.home() / "railiance-cluster" / "tools" / "observe_cluster_resources.py"
|
|
spec = importlib.util.spec_from_file_location("incluster", ROOT / "scripts/weekly_allocation_incluster.py")
|
|
incluster = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(incluster)
|
|
|
|
RESOURCE = "resource:railiance:reef-railiance:k3s"
|
|
FAKE_OBSERVER = '''
|
|
def cpu_millicores(v):
|
|
v = str(v or "0")
|
|
return int(float(v[:-1]) / 1e6) if v.endswith("n") else int(v[:-1]) if v.endswith("m") else int(float(v) * 1000)
|
|
def bytes_value(v):
|
|
v = str(v or "0")
|
|
return int(v[:-2]) * 1024 if v.endswith("Ki") else int(v)
|
|
def build_observation(snapshots, captured_at):
|
|
rows = [{"namespace": p["metadata"]["namespace"], "workload": p["metadata"]["labels"]["app"], "pods": 1,
|
|
"requests": {"cpu_millicores": 10, "memory_bytes": 1}, "limits": {}} for p in snapshots["pods"]["items"]]
|
|
return {"schema_version": "railiance.cluster-resource-observation.v1", "resource_id": "%s",
|
|
"captured_at": captured_at, "utilization": {"workloads": rows},
|
|
"top": {"node": snapshots["node_top"], "pod": snapshots["pod_top"]},
|
|
"measurement_gaps": [] if snapshots["pod_top"] is not None else ["pod metrics unavailable"]}
|
|
''' % RESOURCE
|
|
|
|
|
|
def pod(ns, name, app):
|
|
return {"metadata": {"namespace": ns, "name": name, "labels": {"app": app}},
|
|
"spec": {"nodeName": "railiance01", "containers": [{"name": "c", "resources": {"requests": {"cpu": "10m", "memory": "16Mi"}}}]},
|
|
"status": {"phase": "Running"}}
|
|
|
|
|
|
class InClusterRunnerTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
base = Path(self.tmp.name)
|
|
self.data = base / "data"
|
|
(base / "ownership.json").write_text(json.dumps({
|
|
"host": {"cluster_resource_id": RESOURCE},
|
|
"namespaces": {"pilot": {"owner": "railiance-apps", "service": "pilot", "tenant": "unknown"}}}))
|
|
(base / "inventory.json").write_text(json.dumps({"id": RESOURCE}))
|
|
(base / "observer.py").write_text(FAKE_OBSERVER)
|
|
self.reads, self.queries = [], []
|
|
self.metrics_fail = False
|
|
self.args = argparse.Namespace(
|
|
data_dir=self.data, source_dir=ROOT / "scripts", observer=base / "observer.py",
|
|
queries=ROOT / "config/weekly-allocation-queries.json", ownership=base / "ownership.json",
|
|
inventory=base / "inventory.json", reviews=None, prometheus_url="http://prom.test:9090",
|
|
executor="test", max_age_hours=48.0)
|
|
|
|
def tearDown(self):
|
|
self.tmp.cleanup()
|
|
|
|
def get(self, path):
|
|
self.reads.append(path)
|
|
if path.startswith("/apis/metrics.k8s.io"):
|
|
if self.metrics_fail:
|
|
raise OSError("metrics unavailable")
|
|
if path.endswith("/nodes"):
|
|
return {"items": [{"metadata": {"name": "railiance01"}, "usage": {"cpu": "1500000000n", "memory": "2048Ki"}}]}
|
|
return {"items": [{"metadata": {"namespace": "pilot", "name": "app-1"},
|
|
"containers": [{"usage": {"cpu": "3m", "memory": "1Ki"}}, {"usage": {"cpu": "2m", "memory": "1Ki"}}]}]}
|
|
if path == "/api/v1/pods":
|
|
return {"items": [pod("pilot", "app-1", "app")]}
|
|
return {"items": []}
|
|
|
|
def query(self, promql, at):
|
|
self.queries.append((promql, at))
|
|
return {"status": "success", "data": {"resultType": "vector", "result": [
|
|
{"metric": {"namespace": "pilot"}, "value": [at.timestamp(), "1.5"]}]}}
|
|
|
|
def run_at(self, when):
|
|
return incluster.run(self.args, now=when, get=self.get, query=self.query)
|
|
|
|
def test_week_identity_and_window_end_are_dst_independent(self):
|
|
summer = datetime(2026, 10, 19, 6, 0, 30, tzinfo=timezone.utc) # 08:00 CEST
|
|
winter = datetime(2026, 11, 2, 7, 0, 30, tzinfo=timezone.utc) # 08:00 CET
|
|
self.assertEqual(incluster.week_of(summer), ("2026-W43", datetime(2026, 10, 19, 6, tzinfo=timezone.utc)))
|
|
self.assertEqual(incluster.week_of(winter), ("2026-W45", datetime(2026, 11, 2, 6, tzinfo=timezone.utc)))
|
|
early = datetime(2026, 10, 19, 5, 59, tzinfo=timezone.utc)
|
|
self.assertEqual(incluster.week_of(early)[0], "2026-W42")
|
|
self.assertEqual(incluster.missed_weeks("2026-W41", "2026-W44"), ["2026-W42", "2026-W43"])
|
|
self.assertEqual(incluster.missed_weeks("2026-W52", "2027-W01"), ["2026-W53"]) # 2026 has 53 ISO weeks
|
|
self.assertEqual(incluster.missed_weeks("2026-W53", "2027-W01"), [])
|
|
|
|
def test_reads_only_collector_paths_and_retains_report(self):
|
|
receipt = self.run_at(datetime(2026, 10, 19, 6, 0, 30, tzinfo=timezone.utc))
|
|
self.assertEqual(receipt["status"], "retained")
|
|
self.assertEqual(set(self.reads), set(incluster.API_PATHS.values()) | set(incluster.METRICS_PATHS.values()))
|
|
self.assertFalse(any("secret" in r or "configmap" in r for r in self.reads))
|
|
report = json.loads((self.data / "reports/2026-W43.json").read_text())
|
|
self.assertEqual(report["period"]["capture_end"], "2026-10-19T06:00:00+00:00")
|
|
self.assertEqual(report["workloads"][0]["owner"], "railiance-apps")
|
|
self.assertIn("pilot", report["namespace_context"])
|
|
self.assertTrue(all(at == datetime(2026, 10, 19, 6, tzinfo=timezone.utc) for _, at in self.queries))
|
|
cluster = json.loads((Path(receipt["inputs_dir"]) / "cluster.json").read_text())
|
|
self.assertEqual(cluster["top"]["pod"], "pilot app-1 5m 2048\n")
|
|
self.assertEqual(cluster["top"]["node"], "railiance01 1500m 0% 2097152 0%\n")
|
|
provenance = json.loads((Path(receipt["inputs_dir"]) / "provenance.json").read_text())
|
|
self.assertFalse(provenance["secret_surfaces_read"])
|
|
self.assertEqual(provenance["source_observation"], cluster["captured_at"])
|
|
|
|
def test_retry_in_same_week_is_noop_and_next_week_compares(self):
|
|
first = datetime(2026, 10, 19, 6, 0, 30, tzinfo=timezone.utc)
|
|
self.run_at(first)
|
|
before = len(self.reads)
|
|
again = self.run_at(datetime(2026, 10, 20, 9, 0, tzinfo=timezone.utc))
|
|
self.assertEqual(again["status"], "already-retained")
|
|
self.assertEqual(len(self.reads), before)
|
|
nxt = self.run_at(datetime(2026, 10, 26, 7, 0, 30, tzinfo=timezone.utc))
|
|
self.assertEqual((nxt["previous_week"], nxt["missed_weeks"]), ("2026-W43", []))
|
|
self.assertTrue(nxt["comparison_compatible"])
|
|
later = self.run_at(datetime(2026, 11, 16, 7, 0, 30, tzinfo=timezone.utc))
|
|
self.assertEqual(later["missed_weeks"], ["2026-W45", "2026-W46"])
|
|
self.assertFalse(later["comparison_compatible"])
|
|
|
|
def test_metrics_api_outage_is_a_gap_and_query_errors_are_reported(self):
|
|
self.metrics_fail = True
|
|
self.query = lambda promql, at: {"status": "error", "errorType": "bad_data", "error": "x"}
|
|
receipt = self.run_at(datetime(2026, 10, 19, 6, 0, 30, tzinfo=timezone.utc))
|
|
self.assertEqual(receipt["status"], "retained")
|
|
self.assertIn("cpu_p95_m", receipt["failed_queries"])
|
|
self.assertGreaterEqual(receipt["measurement_gaps"], 1)
|
|
|
|
def test_failure_leaves_failed_receipt_and_no_report(self):
|
|
self.args.ownership = None
|
|
with unittest.mock.patch.object(incluster, "parser") as fake_parser, \
|
|
unittest.mock.patch.object(incluster, "datetime") as fake_dt:
|
|
fake_parser.return_value.parse_args.return_value = self.args
|
|
fake_dt.now.return_value = datetime(2026, 10, 19, 6, 0, 30, tzinfo=timezone.utc)
|
|
self.assertEqual(incluster.main(), 1)
|
|
receipts = [json.loads(l) for l in (self.data / "receipts.jsonl").read_text().splitlines()]
|
|
self.assertEqual(receipts[-1]["status"], "failed")
|
|
self.assertFalse((self.data / "reports").exists())
|
|
|
|
def test_prometheus_http_error_body_is_retained(self):
|
|
import io
|
|
import urllib.error
|
|
|
|
def opener(url, timeout):
|
|
raise urllib.error.HTTPError(url, 422, "x", {}, io.BytesIO(b'{"status":"error","error":"bad"}'))
|
|
self.assertEqual(incluster.prometheus_query("http://p", "up", datetime.now(timezone.utc), opener)["error"], "bad")
|
|
|
|
@unittest.skipUnless(OBSERVER.exists(), "railiance-cluster checkout not present")
|
|
def test_real_collector_accepts_rendered_top_lines(self):
|
|
self.args.observer = OBSERVER
|
|
self.reads.clear()
|
|
observer = incluster.load(OBSERVER, "observer_real")
|
|
snapshots = incluster.collect_snapshots(lambda p: {"items": [{"metadata": {"name": "railiance01"},
|
|
"status": {"allocatable": {"cpu": "4", "memory": "8Gi", "pods": "110"}}}]} if p == "/api/v1/nodes"
|
|
else self.get(p), observer)
|
|
observation = observer.build_observation(snapshots, "2026-10-19T06:00:30Z")
|
|
self.assertEqual(observation["resource_id"], RESOURCE)
|
|
self.assertNotIn("node metrics unavailable", observation["measurement_gaps"])
|
|
self.assertNotIn("pod metrics unavailable", observation["measurement_gaps"])
|
|
self.assertIn('"cpu_millicores": 1500', json.dumps(observation))
|
|
|
|
|
|
import unittest.mock # noqa: E402
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|
|
|
|
|
|
class DeliveryTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.data = Path(self.tmp.name)
|
|
report = {"recommendations": [{"decision": "investigate"}, {"decision": "investigate"}],
|
|
"period": {"capture_end": "2026-10-19T06:00:00+00:00"}, "workloads": [{}, {}],
|
|
"namespace_context": {"a": {}}, "comparison": {"compatible": False},
|
|
"measurement_gaps": [], "queries": {"cpu_p95_m": {"response_status": "success"}}}
|
|
(self.data / "r.json").write_text(json.dumps(report))
|
|
self.receipt = {"week": "2026-W43", "status": "retained", "report": str(self.data / "r.json"),
|
|
"report_sha256": "abc"}
|
|
self.posts = []
|
|
|
|
def tearDown(self):
|
|
self.tmp.cleanup()
|
|
|
|
def post(self, hub, body):
|
|
self.posts.append((hub, body))
|
|
return "msg-1"
|
|
|
|
def test_posts_once_per_week_and_records_id(self):
|
|
receipt, ok = incluster.deliver(self.data, self.receipt, "http://hub", self.post)
|
|
self.assertTrue(ok)
|
|
self.assertEqual((receipt["delivery"], receipt["message_id"]), ("posted", "msg-1"))
|
|
hub, body = self.posts[0]
|
|
self.assertEqual((body["from_agent"], body["to_agent"]), ("custodian-allocation-review", "the-custodian"))
|
|
content = json.loads(body["body"])
|
|
self.assertEqual((content["report_sha256"], content["recommendations"]), ("abc", {"investigate": 2}))
|
|
incluster.append_receipt(self.data, receipt)
|
|
again, ok = incluster.deliver(self.data, {**self.receipt, "status": "already-retained"}, "http://hub", self.post)
|
|
self.assertEqual((again["delivery"], again["message_id"], len(self.posts)), ("already-posted", "msg-1", 1))
|
|
|
|
def test_failed_post_is_retried_by_next_run(self):
|
|
def broken(hub, body):
|
|
raise OSError("down")
|
|
receipt, ok = incluster.deliver(self.data, self.receipt, "http://hub", broken)
|
|
self.assertFalse(ok)
|
|
self.assertEqual(receipt["delivery"], "failed: OSError")
|
|
incluster.append_receipt(self.data, receipt)
|
|
retry, ok = incluster.deliver(self.data, {**self.receipt, "status": "already-retained"}, "http://hub", self.post)
|
|
self.assertTrue(ok)
|
|
self.assertEqual(retry["delivery"], "posted")
|
|
|
|
def test_unconfigured_hub_and_failed_runs_do_not_post(self):
|
|
self.assertEqual(incluster.deliver(self.data, self.receipt, None, self.post)[0]["delivery"], "not configured")
|
|
self.assertEqual(incluster.deliver(self.data, {"status": "failed", "week": "x"}, "http://hub", self.post)[1], True)
|
|
self.assertEqual(self.posts, [])
|