numos/export_state.py
back to source
# SPDX-License-Identifier: MIT
"""Build numos.state from the ops registry plus seed / distro profiles.
Units are derived from the ops registry rather than authored (fleet profile),
so the unit set changes when the registry changes. The content hash is what
makes that drift detectable instead of silent.
Profiles (numos.distro):
fleet — original infra (net/clock/join) + optional op catalogue
self-init — graph-resolved bare distro that completes offline/QEMU boot
"""
import json
import os
from numos import seed
from numos.distro import (
floor_manifest,
get_profile,
kernel_manifest,
profile_graph,
profile_summary,
)
from numos.state import render
from numos.validate import validate
OPS_URL = "http://127.0.0.1:8080/api/ops"
def unit_name_for_op(op_name):
"""op_SwarmExecutor_v1 style name -> stable unit name."""
return "op-" + op_name.lower().replace("_", "-")
def build_state(ops, profile="fleet"):
"""Assemble a validated state dict from op names plus a distro profile."""
distro = get_profile(profile)
units = [dict(u) for u in distro["units"]]
# Op units are a DISPATCH CATALOGUE, not services — fleet profile only.
if distro.get("include_op_units"):
for op_name in sorted(set(ops)):
units.append({
"name": unit_name_for_op(op_name),
"kind": "oneshot",
"restart": "never",
"backoff_ms": 0,
"backoff_max_ms": 0,
"arch_mask": [],
"health_probe": None,
"requires": ["join"],
"after": [],
"exec": "numctl run-op %s" % op_name,
})
state = {
"version": 1,
"arch_targets": seed.ARCH_TARGETS,
"boot_phases": list(distro["boot_phases"]),
"units": sorted(units, key=lambda u: u["name"]),
"health": list(distro["health"]),
}
validate(state)
return state
def fetch_ops(url=OPS_URL):
"""Read op names from a live IntrikataTopology instance."""
import urllib.request
with urllib.request.urlopen(url, timeout=10) as response:
payload = json.loads(response.read().decode("utf-8"))
if isinstance(payload, dict):
if "ops" not in payload:
raise ValueError(
"ops endpoint returned a mapping with no 'ops' key: %r"
% sorted(payload)[:5])
ops = payload["ops"]
else:
ops = payload
if not isinstance(ops, list):
raise ValueError("ops endpoint did not return a list: %r" % type(ops))
return [op["name"] if isinstance(op, dict) else op for op in ops]
def export(ops, out_dir, profile="fleet", arch="x86_64"):
"""Write numos.state, JSON view, and distro floor manifest. Returns state path."""
state = build_state(ops, profile=profile)
if not os.path.isdir(out_dir):
os.makedirs(out_dir)
state_path = os.path.join(out_dir, "numos.state")
with open(state_path, "w", newline="\n") as handle:
handle.write(render(state))
json_path = os.path.join(out_dir, "numos.state.json")
with open(json_path, "w", newline="\n") as handle:
json.dump(state, handle, sort_keys=True, indent=2)
handle.write("\n")
# Packer-facing floor plan (graph-resolved; not free-form busybox fetch).
try:
floor = floor_manifest(arch, profile)
except KeyError as exc:
floor = {
"profile": profile,
"arch": arch,
"error": str(exc),
}
floor_path = os.path.join(out_dir, "distro.floor.json")
with open(floor_path, "w", newline="\n") as handle:
json.dump(floor, handle, sort_keys=True, indent=2)
handle.write("\n")
# Graph-resolved third-party kernel pin for ISO/GENESIS media path.
try:
kernel = kernel_manifest(arch, profile)
except KeyError as exc:
kernel = {
"profile": profile,
"arch": arch,
"error": str(exc),
"product": "third-party-kernel",
}
kernel_path = os.path.join(out_dir, "distro.kernel.json")
with open(kernel_path, "w", newline="\n") as handle:
json.dump(kernel, handle, sort_keys=True, indent=2)
handle.write("\n")
meta_path = os.path.join(out_dir, "distro.profile.json")
with open(meta_path, "w", newline="\n") as handle:
summary = profile_summary(profile)
# State may include fleet op units beyond the static profile unit list.
summary["unit_names"] = sorted(u["name"] for u in state["units"])
summary["phase_names"] = [
p["name"]
for p in sorted(state["boot_phases"], key=lambda p: p["ordinal"])
]
json.dump(summary, handle, sort_keys=True, indent=2)
handle.write("\n")
# Pure graph product: nodes + edges for tooling / docs (not boot input).
graph_path = os.path.join(out_dir, "distro.graph.json")
with open(graph_path, "w", newline="\n") as handle:
json.dump(profile_graph(profile), handle, sort_keys=True, indent=2)
handle.write("\n")
mermaid_path = os.path.join(out_dir, "distro.graph.mmd")
from numos.distro import mermaid_for_profile
with open(mermaid_path, "w", newline="\n") as handle:
handle.write(mermaid_for_profile(profile))
return state_path
def main():
import argparse
parser = argparse.ArgumentParser(description="Export numos.state")
parser.add_argument("--ops-url", default=OPS_URL)
parser.add_argument("--out", default="dist")
parser.add_argument(
"--profile",
default="fleet",
choices=["fleet", "self-init"],
help="distro profile (self-init = complete offline/QEMU boot graph)",
)
parser.add_argument(
"--arch",
default="x86_64",
help="arch for distro.floor.json resolution",
)
parser.add_argument(
"--seed-only",
action="store_true",
help="do not fetch ops; empty op catalogue (implied for self-init)",
)
args = parser.parse_args()
if args.profile == "self-init" or args.seed_only:
ops = []
else:
try:
ops = fetch_ops(args.ops_url)
except Exception as exc:
# Fail soft for offline export: seed infra only.
print("ops fetch failed (%s); exporting seed-only" % exc)
ops = []
path = export(ops, args.out, profile=args.profile, arch=args.arch)
print("wrote %s (profile=%s)" % (path, args.profile))
if __name__ == "__main__":
main()