diff --git a/config/weekly-allocation-queries.json b/config/weekly-allocation-queries.json new file mode 100644 index 0000000..8170e56 --- /dev/null +++ b/config/weekly-allocation-queries.json @@ -0,0 +1,16 @@ +{ + "schema": "custodian.weekly-allocation-queries.v1", + "note": "CUST-WP-0071-T05. Namespace-level aggregates proven on 2026-09-28 against telemetry/telemetry-prometheus. They populate namespace_context only; workload rows stay 'investigate' until workload-labelled queries exist (kube_pod_labels is not exported by the current kube-state-metrics). Changing any query text changes the report query_signature and breaks week-on-week comparability for one week.", + "window": "7d", + "step": "5m", + "queries": { + "cpu_p95_m": "quantile_over_time(0.95, (sum by(namespace) (rate(container_cpu_usage_seconds_total{container!=\"\",container!=\"POD\",namespace!=\"\"}[5m])) * 1000)[7d:5m])", + "cpu_peak_m": "max_over_time((sum by(namespace) (rate(container_cpu_usage_seconds_total{container!=\"\",container!=\"POD\",namespace!=\"\"}[5m])) * 1000)[7d:5m])", + "cpu_sample_count": "count_over_time((sum by(namespace) (rate(container_cpu_usage_seconds_total{container!=\"\",container!=\"POD\",namespace!=\"\"}[5m])) * 1000)[7d:5m])", + "memory_p95_bytes": "quantile_over_time(0.95, (sum by(namespace) (container_memory_working_set_bytes{container!=\"\",container!=\"POD\",namespace!=\"\"}))[7d:5m])", + "memory_peak_bytes": "max_over_time((sum by(namespace) (container_memory_working_set_bytes{container!=\"\",container!=\"POD\",namespace!=\"\"}))[7d:5m])", + "memory_sample_count": "count_over_time((sum by(namespace) (container_memory_working_set_bytes{container!=\"\",container!=\"POD\",namespace!=\"\"}))[7d:5m])", + "cpu_throttle_ratio": "sum by(namespace) (increase(container_cpu_cfs_throttled_periods_total{container!=\"\",container!=\"POD\",namespace!=\"\"}[7d])) / sum by(namespace) (increase(container_cpu_cfs_periods_total{container!=\"\",container!=\"POD\",namespace!=\"\"}[7d]))", + "restarts": "sum by(namespace) (increase(kube_pod_container_status_restarts_total[7d]))" + } +} diff --git a/docs/weekly-allocation-report.md b/docs/weekly-allocation-report.md index 2b18df2..a17c3bc 100644 --- a/docs/weekly-allocation-report.md +++ b/docs/weekly-allocation-report.md @@ -112,3 +112,42 @@ The tests cover historical namespace-only input, missing/zero/NaN, invalid aggregation, query failure, compatible/incompatible periods, stable identities and owner-task links, stale/future evidence, owner review, resource/capture mismatch and retained retry/collision behavior. + +## In-cluster runner (CUST-WP-0071-T05, founder decision 2026-10-21) + +`docs/decisions/2026-10-21-cust-wp-0071-delivery-interview.md` selects an +in-cluster Kubernetes CronJob for T05 and supersedes the Activity Core +placement above. `scripts/weekly_allocation_incluster.py` is the job entry +point. It is standard-library only and reuses the existing owner code +unchanged: railiance-cluster `build_observation`, `reconcile.join` (PyYAML is +now imported only by the reconcile CLI) and this adapter's `build_report` and +`retain`. + +- Reads: the Kubernetes API through the pod ServiceAccount, with get/list on + nodes, pods, persistentvolumeclaims, persistentvolumes, storageclasses and + metrics.k8s.io nodes/pods only. Prometheus is reached through + `PROMETHEUS_URL` with `config/weekly-allocation-queries.json`. +- Week identity: the ISO week of (run time - 6h). The telemetry window ends at + Monday 06:00 UTC, so consecutive windows are exactly seven days apart in CET + and CEST. Prometheus keeps seven days of data, so the run must happen on the Monday. A + 2026-10-21 (Wednesday) read-only dry run against railiance01 retained the + report, but per-namespace CPU sample counts reached at most 1512 of 2016, + because the window start had already expired. +- Retention: `/reports/-W.json` (create-without-replacement), + per-attempt inputs under `/runs///`, and an append-only + `/receipts.jsonl` with status `retained`, `already-retained` or + `failed`, previous week, `missed_weeks`, comparability and failed queries. +- Idempotency: a retained week is a no-op success. A failed attempt leaves no + report, so the Job retry or a manual run collects again. +- Not done: delivery, owner acknowledgment and owner-task creation + (`delivery: not emitted`). Workload rows stay `investigate` because the queries + are namespace-level. kube-state-metrics does not export `kube_pod_labels`, so + no workload-labelled series exist yet. + +Executor manifests (suspended, not registered with ArgoCD) live in +railiance-platform `argocd/platform-addons/custodian-allocation-review/`, with +the draft Application in `argocd/railiance01/drafts/`. + +```sh +python3 -m pytest tests/test_weekly_allocation_incluster.py tests/test_weekly_allocation_report.py -q +``` diff --git a/scripts/reconcile_railiance_allocation.py b/scripts/reconcile_railiance_allocation.py index 0e61dcd..a7ca729 100644 --- a/scripts/reconcile_railiance_allocation.py +++ b/scripts/reconcile_railiance_allocation.py @@ -10,8 +10,6 @@ from datetime import UTC, datetime from pathlib import Path from typing import Any -import yaml - ROOT = Path(__file__).resolve().parents[1] CLUSTER_OBS = Path.home() / "railiance-cluster" / "tools" / "observe_cluster_resources.py" HEADROOM = Path.home() / "state-hub" / "scripts" / "release_headroom_preflight.py" @@ -131,6 +129,8 @@ def main() -> int: observation = json.loads(args.observation.read_text()) else: observation = cluster.collect(args.kubectl) + import yaml # main only: join() stays stdlib-only for the in-cluster runner + owners = yaml.safe_load(OWNERS.read_text()) report = join(observation, owners) preflight = None diff --git a/scripts/weekly_allocation_incluster.py b/scripts/weekly_allocation_incluster.py new file mode 100644 index 0000000..ddfea2c --- /dev/null +++ b/scripts/weekly_allocation_incluster.py @@ -0,0 +1,272 @@ +#!/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. +""" +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 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, + } + 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) + 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("--prometheus-url", default=env("PROMETHEUS_URL", "http://telemetry-prometheus.telemetry.svc:9090")) + 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 + append_receipt(args.data_dir, receipt) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_weekly_allocation_incluster.py b/tests/test_weekly_allocation_incluster.py new file mode 100644 index 0000000..e8af114 --- /dev/null +++ b/tests/test_weekly_allocation_incluster.py @@ -0,0 +1,168 @@ +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()