ops(observatory): install exact recorded-progress agent layers
This commit is contained in:
DCCONSTRUCTIONS
@@ -0,0 +1,308 @@
|
||||
"""Hash-gated observation-only layer for the two existing recorded agents.
|
||||
|
||||
Pack locally, plan read-only on Worker, then apply the exact plan. No downloads,
|
||||
model changes, resource changes or live jobs. Stopped predecessors are retained.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import copy
|
||||
import io
|
||||
import json
|
||||
import tarfile
|
||||
import time
|
||||
from datetime import UTC, datetime
|
||||
from pathlib import Path
|
||||
from urllib.parse import urlencode
|
||||
|
||||
from migrate_claim_transport_v3 import (
|
||||
READINESS,
|
||||
Engine,
|
||||
canonical,
|
||||
require_idle,
|
||||
save,
|
||||
sha,
|
||||
)
|
||||
|
||||
SOURCE_ROOT = "/opt/nodedc/installed-lab/src/k1link/observatory"
|
||||
BEFORE = {
|
||||
"worker_agent.py": "cf81131746ff4c4a3af6b186a852238f2cdde7e66c70aaa8ac1e9a3dc66bde7c",
|
||||
"worker_http_transport.py": "81bdc27f374cd91358f0eb58603e3e1f53e1f3017bf158d269e75affa4567293",
|
||||
"m49_portable_executor.py": "19a1b5bdb9c8dfc6a91af42ebc648d7f525cce95637f8a368c7679148b9666f0",
|
||||
"m49_portable_source.py": "ab178740e5e577d00863cca6d1d1963acad766b3aa6d7a19f9de870953131fca",
|
||||
"portable_worker_runtime.py": (
|
||||
"1c62636a6da24cae0b9e632ccd34108043dab7d2b0aefba7868376e3ca3feaf7"
|
||||
),
|
||||
"installed_lab_package_runner.py": (
|
||||
"43195cc53e524ff57307ead691441cdd7c2ba09949bbdae6fe77920a63263eac"
|
||||
),
|
||||
"recorded_progress.py": None,
|
||||
"m49_timing_progress.py": None,
|
||||
}
|
||||
TARGETS = {
|
||||
"ndc-observatory-m49-worker-agent": (
|
||||
"7aa6ccd2ddba4ebb0c07d793e5331539a2e13b07961c269439d5b79762328194"
|
||||
),
|
||||
"ndc-observatory-installed-lab-worker-agent": (
|
||||
"1b1e335916c1c3d77888b7537331e9d775d078957c4ac0f82e91852301725390"
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def probe() -> str:
|
||||
return f"""import hashlib, json, pathlib
|
||||
from k1link.observatory import worker_agent
|
||||
root=pathlib.Path(worker_agent.__file__).parent
|
||||
assert str(root) == {SOURCE_ROOT!r}
|
||||
print(json.dumps({{name: hashlib.sha256((root/name).read_bytes()).hexdigest()
|
||||
if (root/name).is_file() else None for name in {list(BEFORE)!r}}}))
|
||||
"""
|
||||
|
||||
|
||||
def pack(repository: Path, output: Path) -> None:
|
||||
output.mkdir(parents=False, exist_ok=False)
|
||||
files = {}
|
||||
for name in BEFORE:
|
||||
payload = (repository / "src/k1link/observatory" / name).read_bytes()
|
||||
compile(payload, name, "exec")
|
||||
(output / name).write_bytes(payload)
|
||||
files[name] = sha(payload)
|
||||
save(output / "payload.json", {"schema_version": 1, "files": files})
|
||||
|
||||
|
||||
def payload_files(root: Path) -> dict[str, bytes]:
|
||||
manifest = json.loads((root / "payload.json").read_bytes())
|
||||
if set(manifest) != {"schema_version", "files"} or manifest["schema_version"] != 1:
|
||||
raise ValueError("invalid progress payload manifest")
|
||||
if set(manifest["files"]) != set(BEFORE):
|
||||
raise ValueError("progress file set changed")
|
||||
files = {}
|
||||
for name, digest in manifest["files"].items():
|
||||
path = root / name
|
||||
if path.is_symlink() or not path.is_file() or path.stat().st_size > 256_000:
|
||||
raise ValueError("unsafe progress payload")
|
||||
value = path.read_bytes()
|
||||
if sha(value) != digest:
|
||||
raise ValueError("progress payload hash mismatch")
|
||||
compile(value, name, "exec")
|
||||
files[name] = value
|
||||
return files
|
||||
|
||||
|
||||
def validate_target(name: str, row: dict) -> None:
|
||||
host, config = row["HostConfig"], row["Config"]
|
||||
if row["Name"] != "/" + name or row["Image"] != "sha256:" + TARGETS[name]:
|
||||
raise ValueError("agent identity changed")
|
||||
if not row["State"]["Running"] or not host["ReadonlyRootfs"]:
|
||||
raise ValueError("agent is not in its expected running/read-only state")
|
||||
if host["NetworkMode"] != "bridge" or host.get("DeviceRequests") or host.get("Privileged"):
|
||||
raise ValueError("agent GPU/network/privilege boundary changed")
|
||||
if config["Labels"].get("com.nodedc.authority") != "observation-only":
|
||||
raise ValueError("agent authority changed")
|
||||
for entry in config["Env"]:
|
||||
key = entry.split("=", 1)[0].upper()
|
||||
if any(word in key for word in ("TOKEN", "PASSWORD", "SECRET")) and not key.endswith(
|
||||
"_FILE"
|
||||
):
|
||||
raise ValueError("inline secret in declaration")
|
||||
|
||||
|
||||
def create_hash(row: dict) -> str:
|
||||
return sha(canonical({"Config": row["Config"], "HostConfig": row["HostConfig"]}))
|
||||
|
||||
|
||||
def validate_fence(target: dict, row: dict) -> None:
|
||||
validate_target(target["name"], row)
|
||||
if row["Id"] != target["id"] or create_hash(row) != target["create_sha256"]:
|
||||
raise ValueError("agent or declaration changed since plan")
|
||||
|
||||
|
||||
def plan(engine: Engine, root: Path) -> dict:
|
||||
files = payload_files(root)
|
||||
targets = []
|
||||
for name in TARGETS:
|
||||
row = engine.inspect(name)
|
||||
validate_target(name, row)
|
||||
if engine.execute_json(name, probe()) != BEFORE:
|
||||
raise ValueError("imported agent code is not the reviewed baseline")
|
||||
require_idle(engine.execute_json(name, READINESS))
|
||||
targets.append(
|
||||
{
|
||||
"name": name,
|
||||
"id": row["Id"],
|
||||
"parent": TARGETS[name],
|
||||
"create_sha256": create_hash(row),
|
||||
}
|
||||
)
|
||||
return {
|
||||
"schema_version": "missioncore.recorded-progress-install-plan/v1",
|
||||
"targets": targets,
|
||||
"files": {name: sha(value) for name, value in files.items()},
|
||||
"installer_sha256": sha(Path(__file__).read_bytes()),
|
||||
"engine_helper_sha256": sha(
|
||||
Path(__file__).with_name("migrate_claim_transport_v3.py").read_bytes()
|
||||
),
|
||||
"compute_packages_changed": False,
|
||||
}
|
||||
|
||||
|
||||
def build(engine: Engine, target: dict, files: dict[str, bytes], plan_sha: str) -> str:
|
||||
created = engine.request(
|
||||
"POST",
|
||||
"/containers/create",
|
||||
{
|
||||
"Image": "sha256:" + target["parent"],
|
||||
"Entrypoint": ["/bin/true"],
|
||||
"Cmd": [],
|
||||
"HostConfig": {
|
||||
"NetworkMode": "none",
|
||||
"CapDrop": ["ALL"],
|
||||
"PidsLimit": 32,
|
||||
"SecurityOpt": ["no-new-privileges"],
|
||||
},
|
||||
},
|
||||
)["Id"]
|
||||
try:
|
||||
engine.request("POST", f"/containers/{created}/start")
|
||||
if engine.request("POST", f"/containers/{created}/wait")["StatusCode"] != 0:
|
||||
raise ValueError("offline layer initialization failed")
|
||||
buffer = io.BytesIO()
|
||||
with tarfile.open(fileobj=buffer, mode="w") as archive:
|
||||
for name, payload in files.items():
|
||||
member = tarfile.TarInfo(name)
|
||||
member.size, member.mode, member.mtime = len(payload), 0o644, int(time.time())
|
||||
archive.addfile(member, io.BytesIO(payload))
|
||||
engine.request(
|
||||
"PUT",
|
||||
f"/containers/{created}/archive?" + urlencode({"path": SOURCE_ROOT}),
|
||||
buffer.getvalue(),
|
||||
)
|
||||
changes = engine.request("GET", f"/containers/{created}/changes")
|
||||
allowed = {str(Path(SOURCE_ROOT) / name) for name in files}
|
||||
parents = {str(p) for p in Path(SOURCE_ROOT).parents} | {SOURCE_ROOT}
|
||||
if not changes or any(
|
||||
item["Path"] not in allowed | parents or item["Kind"] not in (0, 1) for item in changes
|
||||
):
|
||||
raise ValueError("unrelated filesystem changes in progress layer")
|
||||
if not allowed.issubset({item["Path"] for item in changes}):
|
||||
raise ValueError("a progress file was omitted from the image layer")
|
||||
parent = engine.request("GET", f"/images/sha256:{target['parent']}/json")
|
||||
config = copy.deepcopy(parent["Config"])
|
||||
config.setdefault("Labels", {}).update(
|
||||
{
|
||||
"com.nodedc.recorded-progress.plan-sha256": plan_sha,
|
||||
"com.nodedc.recorded-progress.parent-sha256": target["parent"],
|
||||
}
|
||||
)
|
||||
image = engine.request(
|
||||
"POST",
|
||||
"/commit?"
|
||||
+ urlencode(
|
||||
{
|
||||
"container": created,
|
||||
"repo": target["name"] + "-progress",
|
||||
"tag": "v1",
|
||||
}
|
||||
),
|
||||
config,
|
||||
)["Id"]
|
||||
result = engine.request("GET", f"/images/{image}/json")
|
||||
if result["RootFS"]["Layers"][:-1] != parent["RootFS"]["Layers"]:
|
||||
raise ValueError("parent image layers changed")
|
||||
return image
|
||||
finally:
|
||||
engine.request("DELETE", f"/containers/{created}")
|
||||
|
||||
|
||||
def apply(engine: Engine, root: Path, expected: str, evidence: Path) -> dict:
|
||||
started_at = datetime.now(UTC).isoformat()
|
||||
started_mono = time.monotonic_ns()
|
||||
proposal = plan(engine, root)
|
||||
if sha(canonical(proposal)) != expected:
|
||||
raise ValueError("progress install plan changed")
|
||||
evidence.mkdir(parents=False, exist_ok=False)
|
||||
save(evidence / "plan.json", proposal)
|
||||
files = payload_files(root)
|
||||
results = []
|
||||
for target in proposal["targets"]:
|
||||
name = target["name"]
|
||||
before = engine.inspect(name)
|
||||
validate_fence(target, before)
|
||||
image = build(engine, target, files, expected)
|
||||
require_idle(engine.execute_json(name, READINESS))
|
||||
validate_fence(target, engine.inspect(name))
|
||||
body = copy.deepcopy(before["Config"])
|
||||
body["Image"] = image
|
||||
body["Labels"]["com.nodedc.recorded-progress.plan-sha256"] = expected
|
||||
body["HostConfig"] = copy.deepcopy(before["HostConfig"])
|
||||
backup = name + "-pre-progress-" + before["Id"][:12]
|
||||
declaration = {
|
||||
"name": name,
|
||||
"create_body": body,
|
||||
"rollback_name": backup,
|
||||
"rollback_container_id": before["Id"],
|
||||
"parent": target["parent"],
|
||||
}
|
||||
save(evidence / (name + "-declaration.json"), declaration)
|
||||
engine.request("POST", f"/containers/{before['Id']}/stop?t=15")
|
||||
engine.request(
|
||||
"POST", f"/containers/{before['Id']}/update", {"RestartPolicy": {"Name": "no"}}
|
||||
)
|
||||
engine.request("POST", f"/containers/{before['Id']}/rename?" + urlencode({"name": backup}))
|
||||
created = engine.request("POST", "/containers/create?" + urlencode({"name": name}), body)[
|
||||
"Id"
|
||||
]
|
||||
engine.request("POST", f"/containers/{created}/start")
|
||||
# A started replacement may already own operator work; never auto-delete it.
|
||||
time.sleep(3)
|
||||
after = engine.inspect(name)
|
||||
if not after["State"]["Running"] or after["RestartCount"] != 0:
|
||||
raise ValueError("replacement needs reconciliation; predecessor retained")
|
||||
if engine.execute_json(name, probe()) != proposal["files"]:
|
||||
raise ValueError("replacement imported another progress payload")
|
||||
results.append(
|
||||
{
|
||||
"name": name,
|
||||
"id": created,
|
||||
"image": image,
|
||||
"rollback": backup,
|
||||
"readiness": engine.execute_json(name, READINESS),
|
||||
}
|
||||
)
|
||||
save(evidence / (name + "-acceptance.json"), results[-1])
|
||||
receipt = {
|
||||
"plan_sha256": expected,
|
||||
"agents": results,
|
||||
"compute_packages_changed": False,
|
||||
"started_at_utc": started_at,
|
||||
"finished_at_utc": datetime.now(UTC).isoformat(),
|
||||
"started_monotonic_ns": started_mono,
|
||||
"finished_monotonic_ns": time.monotonic_ns(),
|
||||
}
|
||||
save(evidence / "receipt.json", receipt)
|
||||
return receipt
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--repository", type=Path)
|
||||
parser.add_argument("--pack", type=Path)
|
||||
parser.add_argument("--payload", type=Path)
|
||||
parser.add_argument("--apply-plan-sha256")
|
||||
parser.add_argument("--evidence", type=Path)
|
||||
args = parser.parse_args()
|
||||
if args.pack:
|
||||
pack(args.repository, args.pack)
|
||||
return
|
||||
engine = Engine()
|
||||
if args.apply_plan_sha256:
|
||||
result = apply(engine, args.payload, args.apply_plan_sha256, args.evidence)
|
||||
else:
|
||||
proposal = plan(engine, args.payload)
|
||||
result = {"plan": proposal, "plan_sha256": sha(canonical(proposal))}
|
||||
print(json.dumps(result, sort_keys=True))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user