From e5f134f35b09fe179e908be0b23e2cfedc4fc4b4 Mon Sep 17 00:00:00 2001 From: mdheller <21163552+mdheller@users.noreply.github.com> Date: Mon, 3 Aug 2026 23:59:35 -0400 Subject: [PATCH] =?UTF-8?q?feat(admission):=20quota=20+=20admission=20plan?= =?UTF-8?q?e=20=E2=80=94=20completes=20the=20Control-Plane=20Agent?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit place() decided WHERE; admission decides WHETHER a subject may run right now, under per-account/per-project budgets — the multi-tenant, cost-governance third of the Control-Plane Agent (placement + quotas + admission). tools/admission.py — AdmissionController: - per-subject quotas (max_concurrent, gpu_max, cost_budget) over a DEFAULT_QUOTA. - admit() fail-closed: over any budget -> denied, with the exceeded dimensions named. - charge()/release() consumption accounting; spend is cumulative (not released). - file-backed ledger (ledger_path) so a CLI enforces a REAL running budget across processes, not a fresh one each invocation. Wired into the spine: executor.run_spine gains admission/admission_key/cost; the check runs BEFORE the Grant is minted, so an over-quota subject never receives a capability (status "denied", nothing placed/granted/dispatched). sourceosctl: `run --cost N` enforces the ledger-backed budget (exit 4 on DENIED, prints usage vs quota); new `quota` subcommand shows a subject's quota + usage. Verified across invocations: two GPU runs, third DENIED at gpu_max, fail-closed. Tests: +9 admission (budgets, charge/release, isolation, ledger persistence, spine integration) = 90 tools tests green. Wired validate REQUIRED, compute-plane CapD link, ledger gitignored. --- .gitignore | 3 + capd/compute-plane.mesh.capd.json | 1 + tools/admission.py | 94 ++++++++++++++++++++++++++ tools/executor.py | 22 ++++-- tools/sourceosctl.py | 28 +++++++- tools/test_admission.py | 107 ++++++++++++++++++++++++++++++ tools/validate.py | 1 + 7 files changed, 247 insertions(+), 9 deletions(-) create mode 100644 tools/admission.py create mode 100644 tools/test_admission.py diff --git a/.gitignore b/.gitignore index 289c95e..3fce6e5 100644 --- a/.gitignore +++ b/.gitignore @@ -21,6 +21,9 @@ artifacts/mesh-heartbeats/*.json # MCP ops-surface receipts (sealed per tool call at runtime; dir kept via .gitkeep) artifacts/mcp-receipts/*.json +# Admission consumption ledger (per-subject usage, runtime state) +artifacts/admission-usage.json + # Porter control-plane build outputs (compiled locally, not committed) artifacts/porter-shim/shim artifacts/cloudshell-hardened-pack/culler/culler diff --git a/capd/compute-plane.mesh.capd.json b/capd/compute-plane.mesh.capd.json index bc2b8ff..fb2ce8b 100644 --- a/capd/compute-plane.mesh.capd.json +++ b/capd/compute-plane.mesh.capd.json @@ -6,6 +6,7 @@ "description": "A user develops on a low-mem box and the same workload scales out, seamlessly, over whatever the mesh offers: a k8s service, an HPC/SLURM supercomputer, WASM at the edge, a p2p/hyperswarm mesh, volunteer compute (BOINC / Folding@home / open-HEP-style), or an RLC-style blockchain compute market. The substrate is chosen by the placement broker from per-project/per-account policy and live mesh availability, scaling out where it can and where volunteer compute is offered. Governed and fail-closed: sensitive workloads never land on an untrusted (volunteer/p2p/blockchain) backend, every placement is sealed, and with no allowed+available backend it falls back to local or blocks rather than shipping work somewhere the policy forbids. Configured in the portal dashboard.", "links": { "broker": "tools/compute_plane.py", + "admission": "tools/admission.py", "executor": "tools/executor.py", "cli": "tools/sourceosctl.py", "telemetry": "tools/mesh_telemetry.py", diff --git a/tools/admission.py b/tools/admission.py new file mode 100644 index 0000000..f37f724 --- /dev/null +++ b/tools/admission.py @@ -0,0 +1,94 @@ +#!/usr/bin/env python3 +"""Quota + admission — the other two thirds of the Control Plane Agent (placement + quotas + +admission). + +`compute_plane.place()` decides WHERE a workload runs. Admission decides WHETHER a subject may run +it at all *right now*, under per-account / per-project budgets: how many jobs concurrently, how many +GPUs, how much spend. This is what makes the platform multi-tenant and cost-governed rather than a +free-for-all scheduler. + +Fail-closed: over budget -> denied, and nothing is placed, granted, or dispatched. Admission is +checked BEFORE the Grant is minted, so an over-quota subject never even receives a capability. On a +successful dispatch the consumption is charged; when a job finishes it is released (spend is +cumulative and not released — you don't get money back). +""" +from __future__ import annotations + +import json +from pathlib import Path + +DEFAULT_QUOTA = {"max_concurrent": 4, "gpu_max": 2, "cost_budget": 100.0} + + +class AdmissionController: + """Per-subject (or per-project) quotas + a live consumption ledger. + + With a `ledger_path`, consumption persists across processes (so a CLI enforces a real running + budget, not a fresh one each invocation).""" + + def __init__(self, quotas: dict[str, dict] | None = None, ledger_path=None): + self._quotas = quotas or {} + self._ledger_path = Path(ledger_path) if ledger_path else None + self._usage: dict[str, dict] = self._load() + + def _load(self) -> dict: + if self._ledger_path and self._ledger_path.exists(): + try: + return json.loads(self._ledger_path.read_text()) + except (OSError, json.JSONDecodeError): + return {} + return {} + + def _save(self) -> None: + if self._ledger_path: + self._ledger_path.parent.mkdir(parents=True, exist_ok=True) + self._ledger_path.write_text(json.dumps(self._usage, sort_keys=True)) + + def quota_for(self, key: str) -> dict: + return {**DEFAULT_QUOTA, **self._quotas.get(key, {})} + + def usage(self, key: str) -> dict: + return self._usage.setdefault(key, {"concurrent": 0, "gpu": 0, "cost": 0.0}) + + def admit(self, key: str, workload: dict, *, cost: float = 1.0) -> dict: + """Fail-closed admission check. Returns {admitted, reason, quota, usage, exceeded}.""" + q = self.quota_for(key) + u = self.usage(key) + gpu = 1 if workload.get("needs_gpu") else 0 + checks = { + "max_concurrent": u["concurrent"] + 1 <= q["max_concurrent"], + "gpu_max": u["gpu"] + gpu <= q["gpu_max"], + "cost_budget": round(u["cost"] + cost, 6) <= q["cost_budget"], + } + exceeded = sorted(name for name, ok in checks.items() if not ok) + return {"admitted": not exceeded, + "reason": "within budget" if not exceeded else f"quota exceeded: {', '.join(exceeded)}", + "quota": q, "usage": dict(u), "exceeded": exceeded} + + def charge(self, key: str, workload: dict, *, cost: float = 1.0) -> dict: + """Record a dispatch against the subject's budget.""" + u = self.usage(key) + u["concurrent"] += 1 + u["gpu"] += 1 if workload.get("needs_gpu") else 0 + u["cost"] = round(u["cost"] + cost, 6) + self._save() + return dict(u) + + def release(self, key: str, workload: dict) -> dict: + """A job finished: free the concurrency + GPU it held. Spend stays spent.""" + u = self.usage(key) + u["concurrent"] = max(0, u["concurrent"] - 1) + u["gpu"] = max(0, u["gpu"] - (1 if workload.get("needs_gpu") else 0)) + self._save() + return dict(u) + + +if __name__ == "__main__": + import json + ac = AdmissionController({"spiffe://sourceos/agent/dev": {"gpu_max": 1, "max_concurrent": 2}}) + key = "spiffe://sourceos/agent/dev" + gpu_wl = {"needs_gpu": True} + print(json.dumps(ac.admit(key, gpu_wl), indent=2)) + ac.charge(key, gpu_wl) + # a second GPU job now exceeds gpu_max=1 + print(json.dumps(ac.admit(key, gpu_wl), indent=2)) diff --git a/tools/executor.py b/tools/executor.py index 4d033c9..2eec5eb 100644 --- a/tools/executor.py +++ b/tools/executor.py @@ -146,13 +146,20 @@ def execute(workload: dict, decision: dict, grant: dict, *, session_id: str, ver def run_spine(workload: dict, policy: dict, *, registry, binding: dict, capability: dict, - attestation: dict, constraints: dict, signer, verifier, apply: bool = False) -> dict: - """The one call that runs a workload governed across the mesh: place -> grant -> verify -> execute. + attestation: dict, constraints: dict, signer, verifier, apply: bool = False, + admission=None, admission_key: str | None = None, cost: float = 1.0) -> dict: + """The one call that runs a workload governed across the mesh: admit -> place -> grant -> verify + -> execute -> charge. - Returns the full trace. If the plane blocks placement (fail-closed), no Grant is issued and - nothing is dispatched. + Returns the full trace. Fail-closed at every stage: over quota -> denied (no Grant minted); the + plane blocks placement -> blocked; neither issues a Grant nor dispatches anything. """ import compute_plane as cp + akey = admission_key or binding["spiffe_id"] + if admission is not None: + adm = admission.admit(akey, workload, cost=cost) + if not adm["admitted"]: + return {"status": "denied", "admission": adm} decision = cp.place(workload, policy, registry.availability()) if not decision.get("backend"): return {"status": "blocked", "decision": decision} @@ -160,8 +167,11 @@ def run_spine(workload: dict, policy: dict, *, registry, binding: dict, capabili attestation=attestation, constraints=constraints, signer=signer) execution = execute(workload, decision, grant, session_id=binding["session_id"], verifier=verifier, apply=apply) - return {"status": "ran", "backend": decision["backend"], "decision": decision, - "grant_id": grant["grant_id"], "execution": execution} + result = {"status": "ran", "backend": decision["backend"], "decision": decision, + "grant_id": grant["grant_id"], "execution": execution} + if admission is not None: + result["admission"] = admission.charge(akey, workload, cost=cost) + return result if __name__ == "__main__": diff --git a/tools/sourceosctl.py b/tools/sourceosctl.py index 6e39473..0c4652b 100644 --- a/tools/sourceosctl.py +++ b/tools/sourceosctl.py @@ -25,6 +25,7 @@ from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) +import admission as adm # noqa: E402 import compute_plane as cp # noqa: E402 import executor as ex # noqa: E402 import mcp_a2a_grant as g # noqa: E402 @@ -32,6 +33,7 @@ ROOT = Path(__file__).resolve().parent.parent HEARTBEATS = ROOT / "artifacts" / "mesh-heartbeats" +LEDGER = ROOT / "artifacts" / "admission-usage.json" _DEV_KEY = "dev-only-key-not-for-production" @@ -48,7 +50,7 @@ def _sha(s: str) -> str: def run_workload(*, name, command, effect, sensitivity, scalable, gpu, image, subject, - heartbeats_dir, key, apply=False, dry=False) -> dict: + heartbeats_dir, key, apply=False, dry=False, admission=None, cost=1.0) -> dict: """Core of `run` — testable without the CLI. Returns the full spine trace (or a placement).""" reg = mt.MeshRegistry.from_dir(heartbeats_dir) workload = {"name": name, "command": command, "effect": effect, "sensitivity": sensitivity, @@ -63,17 +65,23 @@ def run_workload(*, name, command, effect, sensitivity, scalable, gpu, image, su attestation = g.attestation_bundle(spiffe_id=subject, aum_digest=aum, tpm_valid=True, cosign_valid=True) return ex.run_spine(workload, policy, registry=reg, binding=binding, capability=capability, attestation=attestation, constraints={"ops_allow": ["exec.run"]}, - signer=g.hmac_signer(key), verifier=g.hmac_verifier(key), apply=apply) + signer=g.hmac_signer(key), verifier=g.hmac_verifier(key), apply=apply, + admission=admission, admission_key=subject, cost=cost) def cmd_run(args) -> int: if _dev_mode(): print("! DEV MODE: no SOURCEOS_SIGNING_KEY set — synthesizing a dev attestation + HMAC key. " "Not for production.", file=sys.stderr) + admission = None if args.dry else adm.AdmissionController(ledger_path=LEDGER) out = run_workload(name=args.name, command=args.command, effect=args.effect, sensitivity=args.sensitivity, scalable=not args.no_scale, gpu=args.gpu, image=args.image, subject=args.subject, heartbeats_dir=HEARTBEATS, - key=_key(), apply=args.apply, dry=args.dry) + key=_key(), apply=args.apply, dry=args.dry, admission=admission, cost=args.cost) + if out["status"] == "denied": + a = out["admission"] + print(f"DENIED (fail-closed): {a['reason']} [usage {a['usage']} vs quota {a['quota']}]") + return 4 if out["status"] == "blocked": print(f"BLOCKED (fail-closed): {out['decision']['reason']}") return 3 @@ -85,6 +93,15 @@ def cmd_run(args) -> int: print(f"ran on: {out['backend']} ({d['backend_trust']}) via grant {out['grant_id']}") print(f"dispatch: {e['dispatch']['kind']} (applied={e['dispatch'].get('applied')})") print(f"sealed receipt: {e['receipt']['receipt_digest']}") + if out.get("admission"): + print(f"quota usage: {out['admission']}") + return 0 + + +def cmd_quota(args) -> int: + ac = adm.AdmissionController(ledger_path=LEDGER) + print(f"quota for {args.subject}: {ac.quota_for(args.subject)}") + print(f"usage: {ac.usage(args.subject)}") return 0 @@ -125,6 +142,7 @@ def build_parser() -> argparse.ArgumentParser: r.add_argument("--gpu", action="store_true", help="workload needs a GPU") r.add_argument("--no-scale", action="store_true", help="keep it small (non-scalable)") r.add_argument("--subject", default="spiffe://sourceos/agent/dev", help="the requesting subject SPIFFE id") + r.add_argument("--cost", type=float, default=1.0, help="cost units to charge against the subject's budget") r.add_argument("--apply", action="store_true", help="actually execute (local subprocess / kubectl apply)") r.add_argument("--dry", action="store_true", help="only decide placement; dispatch nothing") r.set_defaults(func=cmd_run) @@ -144,6 +162,10 @@ def build_parser() -> argparse.ArgumentParser: c = sub.add_parser("commons", help="the reproducible knowledge commons") c.set_defaults(func=cmd_commons) + + q = sub.add_parser("quota", help="show a subject's quota + current usage") + q.add_argument("--subject", default="spiffe://sourceos/agent/dev") + q.set_defaults(func=cmd_quota) return p diff --git a/tools/test_admission.py b/tools/test_admission.py new file mode 100644 index 0000000..f273931 --- /dev/null +++ b/tools/test_admission.py @@ -0,0 +1,107 @@ +#!/usr/bin/env python3 +"""Tests for quota + admission. Load-bearing: fail-closed over budget (concurrent / gpu / cost), +correct charge/release accounting, per-subject isolation, and that the spine actually refuses to +place/grant/dispatch an over-quota subject.""" +import admission as adm +import executor as ex +import mcp_a2a_grant as g +import mesh_telemetry as mt + +KEY = b"adm-test-key" +DEV = "spiffe://sourceos/agent/dev" + + +def test_admits_within_budget(): + ac = adm.AdmissionController() + r = ac.admit(DEV, {"needs_gpu": True}, cost=1.0) + assert r["admitted"] and r["exceeded"] == [] + + +def test_denies_over_concurrency(): + ac = adm.AdmissionController({DEV: {"max_concurrent": 1}}) + ac.charge(DEV, {}) + r = ac.admit(DEV, {}) + assert not r["admitted"] and "max_concurrent" in r["exceeded"] + + +def test_denies_over_gpu(): + ac = adm.AdmissionController({DEV: {"gpu_max": 1}}) + ac.charge(DEV, {"needs_gpu": True}) + r = ac.admit(DEV, {"needs_gpu": True}) + assert not r["admitted"] and "gpu_max" in r["exceeded"] + + +def test_denies_over_cost_budget(): + ac = adm.AdmissionController({DEV: {"cost_budget": 5.0}}) + ac.charge(DEV, {}, cost=4.5) + r = ac.admit(DEV, {}, cost=1.0) # 4.5 + 1.0 = 5.5 > 5.0 + assert not r["admitted"] and "cost_budget" in r["exceeded"] + + +def test_release_frees_concurrency_and_gpu_but_not_spend(): + ac = adm.AdmissionController() + ac.charge(DEV, {"needs_gpu": True}, cost=3.0) + ac.release(DEV, {"needs_gpu": True}) + u = ac.usage(DEV) + assert u["concurrent"] == 0 and u["gpu"] == 0 and u["cost"] == 3.0 # spend stays spent + + +def test_subjects_have_isolated_budgets(): + ac = adm.AdmissionController({DEV: {"max_concurrent": 1}}) + ac.charge(DEV, {}) + other = "spiffe://sourceos/agent/other" + assert ac.admit(DEV, {})["admitted"] is False # dev is full + assert ac.admit(other, {})["admitted"] is True # other is independent + + +def test_default_quota_applies_to_unknown_subject(): + ac = adm.AdmissionController() + assert ac.quota_for("spiffe://who/dis") == adm.DEFAULT_QUOTA + + +def test_ledger_persists_consumption_across_controllers(): + import pathlib + import tempfile + with tempfile.TemporaryDirectory() as td: + p = pathlib.Path(td) / "usage.json" + ac1 = adm.AdmissionController({DEV: {"max_concurrent": 1}}, ledger_path=p) + ac1.charge(DEV, {}) + ac2 = adm.AdmissionController({DEV: {"max_concurrent": 1}}, ledger_path=p) # fresh process + assert ac2.admit(DEV, {})["admitted"] is False # sees ac1's charge on disk + + +# ── integration with the spine ─────────────────────────────────────────────────────── +def _spine_args(gpu): + aum = "sha256:" + "ab" * 32 + reg = mt.MeshRegistry() + reg.heartbeat("slurm", "hpc-slurm", 100) + return dict( + workload={"name": "t", "sensitivity": "normal", "scalable": True, "needs_gpu": gpu, + "effect": "compute", "command": "echo hi"}, + policy={}, registry=reg, + binding={"spiffe_id": DEV, "aum_digest": aum, "session_id": "sess_adm01"}, + capability={"kind": "mcp_tool", "capability_ref": "capd://caps.x", + "capability_digest": "sha256:" + "cd" * 32, "effect": "compute"}, + attestation=g.attestation_bundle(spiffe_id=DEV, aum_digest=aum, tpm_valid=True, cosign_valid=True), + constraints={"ops_allow": ["exec.run"]}, + signer=g.hmac_signer(KEY), verifier=g.hmac_verifier(KEY)) + + +def test_spine_denies_over_quota_before_granting(): + ac = adm.AdmissionController({DEV: {"gpu_max": 1}}) + args = _spine_args(gpu=True) + first = ex.run_spine(**args, admission=ac, cost=1.0) + assert first["status"] == "ran" and first["admission"]["gpu"] == 1 + second = ex.run_spine(**_spine_args(gpu=True), admission=ac, cost=1.0) + assert second["status"] == "denied" # over gpu_max + assert "grant_id" not in second # no Grant minted, nothing dispatched + assert "gpu_max" in second["admission"]["exceeded"] + + +if __name__ == "__main__": + import sys + fns = [v for k, v in sorted(globals().items()) if k.startswith("test_")] + for fn in fns: + fn() + print(f"ok: {len(fns)} admission tests passed") + sys.exit(0) diff --git a/tools/validate.py b/tools/validate.py index 1103dbc..d40ef18 100644 --- a/tools/validate.py +++ b/tools/validate.py @@ -31,6 +31,7 @@ "tools/mcp_ops_server.py", "tools/executor.py", "tools/sourceosctl.py", + "tools/admission.py", ] CAPD_KEYS = ("capability_id", "kind", "status", "links", "composes_with", "policy") # Every CapD in capd/ must carry the core keys and parse — not just the flagship control-plane one.