diff --git a/scripts/rpctl/README.md b/scripts/rpctl/README.md index edfa69d..55bee88 100644 --- a/scripts/rpctl/README.md +++ b/scripts/rpctl/README.md @@ -1,18 +1,22 @@ # rpctl -`rpctl` 是四款 RP 本地控制 CLI,用于在实验服务器本地独立启动、停止和观察单个 RP/profile 的 all5 实验。一个实验可包含一次 snapshot 和多次 delta 子 run。 +`rpctl` 是五款 RP 本地控制 CLI,用于在实验服务器本地独立启动、停止和观察单个 RP/profile 的 all5 实验。一个实验可包含一次 snapshot 和多次复用 state 的子 run。 ## 部署 ```bash -mkdir -p /root/rpki_4rp_ctl -rsync -a rpki_2/rpki/scripts/rpctl/ root@47.77.237.41:/root/rpki_4rp_ctl/ -rsync -a specs/develop/20260707_3/feature100_four_rp_interval_archive/runtime_bundle/ root@47.77.237.41:/root/rpki_4rp_ctl_runtime_bundle/ -ssh root@47.77.237.41 'cd /root/rpki_4rp_ctl && ./rpctl init --bundle /root/rpki_4rp_ctl_runtime_bundle' +mkdir -p /data/rpki_4rp_ctl +rsync -a rpki_2/rpki/scripts/rpctl/ root@47.251.127.231:/data/rpki_4rp_ctl/ +rsync -a artifacts/rpki-prover/rpki-prover-6857d4bf-linux-amd64.docker.tar.gz root@47.251.127.231:/data/ +ssh root@47.251.127.231 'cd /data/rpki_4rp_ctl && ./rpctl init \ + --bundle /data/rpki_4rp_ctl_runtime_bundle \ + --rpki-prover-image /data/rpki-prover-6857d4bf-linux-amd64.docker.tar.gz' ``` `--bundle` 目录需要包含 #100 runtime bundle 的 `bin/`、`lib/`、`fixtures/`、`scripts/` 和 `versions.json`。不要使用早期 precheck 目录作为标准 bundle,因为它可能缺少 FORT 运行所需动态库。 +`--rpki-prover-image` 是可选项;不配置时原有四款 RP 仍可运行,但启动 rpki-prover 会明确报错。配置后 `init` 会复制并加载离线镜像,校验宿主机和镜像均为 `amd64`,确认 tag 为 `rpki-prover:6857d4bf`、版本为 `rpki-prover-0.10.1`,并将 image ID 与 archive SHA-256 写入 `config.json`。运行时不会在线拉取镜像。 + ## 常用命令 ```bash @@ -20,6 +24,8 @@ ssh root@47.77.237.41 'cd /root/rpki_4rp_ctl && ./rpctl init --bundle /root/rpki ./rpctl status ./rpctl start --rp ours-pp-object-cache --interval 10m --runs 6 --rirs all5 ./rpctl start --rp routinator --interval 0 --runs 1 --rirs apnic +./rpctl start --rp rpki-prover --interval 0 --runs 2 --rirs all5 +./rpctl start --rp rpki-prover-full --interval 0 --runs 1 --rirs all5 ./rpctl continue --exp-id --runs 3 --interval 10m ./rpctl tail ./rpctl show @@ -40,6 +46,10 @@ ssh root@47.77.237.41 'cd /root/rpki_4rp_ctl && ./rpctl init --bundle /root/rpki | `routinator` | `routinator-latest-release` | | `rpki-client` | `rpki-client-latest-release` | | `fort` | `fort-latest-release` | +| `rpki-prover`, `rpki-prover-incremental` | `rpki-prover-incremental` | +| `rpki-prover-full`, `rpki-prover-no-incremental` | `rpki-prover-full` | + +`rpki-prover-incremental` 使用上游默认增量算法;`rpki-prover-full` 显式传入 `--no-incremental-validation`。两者的 state 位于各自实验/profile 目录,互不复用。首轮显示为 `snapshot`;后续和 `continue` 追加轮显示为 `warm`,表示保留了上轮 state,实际 RRDP snapshot/delta 决策仍由 rpki-prover 自身完成。 ## 数据目录 @@ -53,7 +63,9 @@ ssh root@47.77.237.41 'cd /root/rpki_4rp_ctl && ./rpctl init --bundle /root/rpki runs///profiles//runs/run_0001/ ``` -每个子 run 目录包含 `stdout.log`、`stderr.log`、`process-time.txt`、`exit-code.txt`、`run-meta.json` 以及各 RP 产物。 +每个子 run 目录包含 `stdout.log`、`stderr.log`、`exit-code.txt`、`run-meta.json` 以及各 RP 产物。传统四 RP 使用 `process-time.txt`;rpki-prover 额外保存 `container-inspect.json`、`rss-samples.tsv` 和 `vrps.csv`。 + +rpki-prover 的 `run-meta.json` 和 `view` 同时记录 VRP 原始行数与按 `(ASN, prefix, maxLength)` 去重数。其当前 CLI 不输出 VAP,所以统一显示 `n/a`,不是 `0`。max RSS 是容器全部相关 PID 的 RSS 峰值汇总,不是短暂 Docker client 进程的 RSS。 ## 历史查看与删除 diff --git a/scripts/rpctl/rpctl b/scripts/rpctl/rpctl index ab10b41..39dcfba 100755 --- a/scripts/rpctl/rpctl +++ b/scripts/rpctl/rpctl @@ -4,8 +4,10 @@ from __future__ import annotations import argparse import csv import gzip +import hashlib import json import os +import platform import re import shutil import signal @@ -37,6 +39,8 @@ PROFILES = { "routinator-latest-release": "routinator", "rpki-client-latest-release": "rpki-client", "fort-latest-release": "fort", + "rpki-prover-incremental": "rpki-prover", + "rpki-prover-full": "rpki-prover", } ALIASES = { @@ -52,8 +56,15 @@ ALIASES = { "rpki-client-latest-release": "rpki-client-latest-release", "fort": "fort-latest-release", "fort-latest-release": "fort-latest-release", + "rpki-prover": "rpki-prover-incremental", + "rpki-prover-incremental": "rpki-prover-incremental", + "rpki-prover-full": "rpki-prover-full", + "rpki-prover-no-incremental": "rpki-prover-full", } +RPKI_PROVER_IMAGE_TAG = "rpki-prover:6857d4bf" +RPKI_PROVER_VERSION = "rpki-prover-0.10.1" + DEFAULT_CONFIG = { "root": "/root/rpki_4rp_ctl", "bundle": "", @@ -207,6 +218,14 @@ def command_string(argv: list[str]) -> str: return " ".join(shlex.quote(x) for x in argv) +def sha256_file(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + def check_bundle(root: Path) -> None: required = ["bin/rpki", "bin/routinator", "bin/rpki-client", "bin/fort", "fixtures/tal", "scripts/normalize_and_collect.py", "scripts/write_group_report.py", "versions.json"] missing = [item for item in required if not (root / item).exists()] @@ -232,6 +251,102 @@ def install_from_bundle(bundle: Path) -> None: item.chmod(item.stat().st_mode | 0o111) +def validate_rpki_prover_host() -> None: + arch = platform.machine().lower() + if arch not in {"x86_64", "amd64"}: + raise SystemExit(f"rpki-prover image requires amd64 host, found: {arch}") + if shutil.which("docker") is None: + raise SystemExit("docker is required for rpki-prover profiles") + result = subprocess.run( + ["docker", "info"], + stdout=subprocess.DEVNULL, + stderr=subprocess.PIPE, + text=True, + check=False, + ) + if result.returncode != 0: + raise SystemExit(f"docker is not available: {result.stderr.strip()}") + + +def inspect_rpki_prover_image() -> dict[str, str]: + inspect = subprocess.run( + ["docker", "image", "inspect", RPKI_PROVER_IMAGE_TAG], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + check=False, + ) + if inspect.returncode != 0: + raise SystemExit(f"loaded image is missing tag {RPKI_PROVER_IMAGE_TAG}: {inspect.stderr.strip()}") + data = json.loads(inspect.stdout)[0] + architecture = str(data.get("Architecture") or "") + if architecture != "amd64": + raise SystemExit(f"rpki-prover image architecture mismatch: expected amd64, found {architecture or ''}") + version = subprocess.run( + ["docker", "run", "--rm", "--entrypoint", "/opt/app/rpki-prover", RPKI_PROVER_IMAGE_TAG, "--version"], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + check=False, + ) + version_text = version.stdout.strip() + if version.returncode != 0 or version_text != RPKI_PROVER_VERSION: + raise SystemExit( + f"rpki-prover version check failed: expected {RPKI_PROVER_VERSION}, " + f"exit={version.returncode} output={version_text or version.stderr.strip()}" + ) + return { + "tag": RPKI_PROVER_IMAGE_TAG, + "imageId": str(data.get("Id") or ""), + "architecture": architecture, + "version": version_text, + } + + +def install_rpki_prover_image(archive: Path) -> dict[str, str]: + if not archive.is_file(): + raise SystemExit(f"rpki-prover image archive does not exist: {archive}") + validate_rpki_prover_host() + image_dir = ctl_root() / "images" + image_dir.mkdir(parents=True, exist_ok=True) + destination = image_dir / archive.name + if archive != destination: + shutil.copy2(archive, destination) + load = subprocess.run( + ["docker", "load", "--input", str(destination)], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + check=False, + ) + if load.returncode != 0: + raise SystemExit(f"failed to load rpki-prover image archive: {load.stderr.strip()}") + metadata = inspect_rpki_prover_image() + metadata.update({ + "archive": str(destination), + "archiveSha256": sha256_file(destination), + "loadedAtUtc": utc_now(), + }) + return metadata + + +def require_rpki_prover_image() -> dict[str, Any]: + metadata = load_config().get("rpkiProverImage") + if not isinstance(metadata, dict) or metadata.get("tag") != RPKI_PROVER_IMAGE_TAG: + raise SystemExit( + "rpki-prover image is not initialized; run ./rpctl init --bundle " + "--rpki-prover-image " + ) + validate_rpki_prover_host() + inspected = inspect_rpki_prover_image() + if metadata.get("imageId") and metadata.get("imageId") != inspected.get("imageId"): + raise SystemExit( + f"rpki-prover image id changed: configured={metadata.get('imageId')} " + f"loaded={inspected.get('imageId')}" + ) + return metadata + + def cmd_init(args: argparse.Namespace) -> None: ensure_dirs() bundle = Path(args.bundle).resolve() @@ -239,9 +354,16 @@ def cmd_init(args: argparse.Namespace) -> None: cfg = load_config() cfg["bundle"] = str(bundle) cfg["initializedAtUtc"] = utc_now() + if args.rpki_prover_image: + cfg["rpkiProverImage"] = install_rpki_prover_image(Path(args.rpki_prover_image).resolve()) save_config(cfg) print(f"initialized root={ctl_root()}") print(f"bundle={bundle}") + if cfg.get("rpkiProverImage"): + image = cfg["rpkiProverImage"] + print(f"rpki_prover_image={image.get('tag')}") + print(f"rpki_prover_image_id={image.get('imageId')}") + print(f"rpki_prover_archive_sha256={image.get('archiveSha256')}") def timed_run(run_dir: Path, argv: list[str], env: dict[str, str] | None = None) -> int: @@ -396,6 +518,240 @@ def build_fort_args(run_dir: Path, state: Path, rirs: list[str]) -> tuple[list[s return argv, env +def prepare_rpki_prover_state(state: Path, rirs: list[str]) -> None: + tal_dir = state / "tals" + tal_dir.mkdir(parents=True, exist_ok=True) + for old_tal in tal_dir.glob("*.tal"): + old_tal.unlink() + for rir in rirs: + shutil.copy2(ctl_root() / "fixtures" / "tal" / RIR_TAL[rir], tal_dir / RIR_TAL[rir]) + + +def rpki_prover_arguments(profile: str) -> list[str]: + argv = [ + "--once", + "--vrp-output", "/outputs/vrps.csv", + "--no-rir-tals", + "--rpki-root-directory", "/rpki-data", + "--cpu-count", "4", + "--fetcher-count", "8", + "--log-level", "info", + ] + if profile == "rpki-prover-full": + argv.append("--no-incremental-validation") + return argv + + +def build_rpki_prover_command(profile: str, run_dir: Path, state: Path, container_name: str) -> list[str]: + return [ + "docker", "run", "--name", container_name, + "-v", f"{state.resolve()}:/rpki-data", + "-v", f"{run_dir.resolve()}:/outputs", + RPKI_PROVER_IMAGE_TAG, + *rpki_prover_arguments(profile), + ] + + +def container_rss_kb(container_name: str) -> tuple[int, int]: + top = subprocess.run( + ["docker", "top", container_name, "-eo", "pid"], + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + text=True, + check=False, + ) + total = 0 + maximum = 0 + for line in top.stdout.splitlines(): + pid = line.strip() + if not pid.isdigit(): + continue + try: + status = Path(f"/proc/{pid}/status").read_text(encoding="utf-8", errors="replace") + except OSError: + continue + match = re.search(r"^VmRSS:\s+(\d+)\s+kB$", status, re.MULTILINE) + if not match: + continue + rss = int(match.group(1)) + total += rss + maximum = max(maximum, rss) + return total, maximum + + +def timed_rpki_prover_run( + profile: str, + run_dir: Path, + state: Path, + container_name: str, +) -> tuple[int, int, int, int, dict[str, Any]]: + run_dir.mkdir(parents=True, exist_ok=True) + start_utc = utc_now() + start_ns = time.monotonic_ns() + (run_dir / "start-utc.txt").write_text(start_utc + "\n", encoding="utf-8") + create_argv = [ + "docker", "create", "--name", container_name, + "-v", f"{state.resolve()}:/rpki-data", + "-v", f"{run_dir.resolve()}:/outputs", + RPKI_PROVER_IMAGE_TAG, + *rpki_prover_arguments(profile), + ] + create = subprocess.run(create_argv, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=False) + if create.returncode != 0: + (run_dir / "stderr.log").write_text(create.stderr, encoding="utf-8") + (run_dir / "stdout.log").write_text(create.stdout, encoding="utf-8") + (run_dir / "end-utc.txt").write_text(utc_now() + "\n", encoding="utf-8") + (run_dir / "exit-code.txt").write_text(str(create.returncode) + "\n", encoding="utf-8") + return create.returncode, 0, 0, (time.monotonic_ns() - start_ns) // 1_000_000, {} + + previous_sigterm = signal.getsignal(signal.SIGTERM) + + def terminate_container(signum: int, _frame: Any) -> None: + subprocess.run( + ["docker", "rm", "--force", container_name], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + check=False, + ) + raise SystemExit(128 + signum) + + signal.signal(signal.SIGTERM, terminate_container) + + max_total_rss_kb = 0 + max_process_rss_kb = 0 + inspect_data: dict[str, Any] = {} + samples_path = run_dir / "rss-samples.tsv" + with (run_dir / "stdout.log").open("wb") as stdout, (run_dir / "stderr.log").open("wb") as stderr, samples_path.open("w", encoding="utf-8") as samples: + samples.write("epoch_ms\ttotal_rss_kb\tmax_process_rss_kb\n") + start = subprocess.Popen(["docker", "start", "--attach", container_name], stdout=stdout, stderr=stderr) + while start.poll() is None: + total_rss_kb, process_rss_kb = container_rss_kb(container_name) + max_total_rss_kb = max(max_total_rss_kb, total_rss_kb) + max_process_rss_kb = max(max_process_rss_kb, process_rss_kb) + samples.write(f"{int(time.time() * 1000)}\t{total_rss_kb}\t{process_rss_kb}\n") + samples.flush() + time.sleep(0.25) + start.wait() + + try: + inspect = subprocess.run( + ["docker", "inspect", container_name], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + check=False, + ) + if inspect.returncode == 0: + parsed = json.loads(inspect.stdout) + if parsed: + inspect_data = parsed[0] + write_json(run_dir / "container-inspect.json", inspect_data) + state_data = inspect_data.get("State", {}) if inspect_data else {} + exit_code = int(state_data.get("ExitCode", start.returncode if start.returncode is not None else 1)) + finally: + subprocess.run(["docker", "rm", "--force", container_name], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False) + signal.signal(signal.SIGTERM, previous_sigterm) + wall_ms = (time.monotonic_ns() - start_ns) // 1_000_000 + (run_dir / "end-utc.txt").write_text(utc_now() + "\n", encoding="utf-8") + (run_dir / "exit-code.txt").write_text(str(exit_code) + "\n", encoding="utf-8") + return exit_code, max_total_rss_kb, max_process_rss_kb, wall_ms, inspect_data + + +def collect_rpki_prover_vrps(path: Path) -> tuple[int, set[str]]: + unique: set[str] = set() + raw = 0 + if not path.is_file(): + return raw, unique + with path.open("r", encoding="utf-8", errors="replace", newline="") as handle: + for row in csv.reader(handle): + if not row or row[0].strip().lower() == "asn": + continue + if len(row) < 3: + continue + raw += 1 + unique.add(",".join(value.strip() for value in row[:3])) + return raw, unique + + +def write_rpki_prover_run_metadata( + run_root: Path, + profile: str, + run_seq: int, + mode: str, + wall_ms: int, + max_rss_kb: int, + max_process_rss_kb: int, + exit_code: int, +) -> dict[str, Any]: + run_dir = run_root / "profiles" / profile / "runs" / f"run_{run_seq:04d}" + raw_vrps, unique_vrps = collect_rpki_prover_vrps(run_dir / "vrps.csv") + with gzip.open(run_dir / "normalized-vrps.txt.gz", "wt", encoding="utf-8") as handle: + for row in sorted(unique_vrps): + handle.write(row + "\n") + image = load_config().get("rpkiProverImage", {}) + meta = { + "profile": profile, + "rp": "rpki-prover", + "runSeq": run_seq, + "syncMode": mode, + "startUtc": read_text_strip(run_dir / "start-utc.txt"), + "endUtc": read_text_strip(run_dir / "end-utc.txt"), + "exitCode": exit_code, + "timing": { + "wallMs": wall_ms, + "maxRssKb": max_rss_kb, + "maxProcessRssKb": max_process_rss_kb, + }, + "counts": { + "vrps": raw_vrps, + "vrpsRaw": raw_vrps, + "vrpsUnique": len(unique_vrps), + "vaps": "n/a", + }, + "vapsAvailable": False, + "validationAlgorithm": "full-every-iteration" if profile == "rpki-prover-full" else "incremental", + "image": image, + "remoteRunDir": str(run_dir), + } + write_json(run_dir / "run-meta.json", meta) + return meta + + +def write_rpki_prover_group_report(run_root: Path, profile: str) -> None: + rows: list[dict[str, Any]] = [] + run_dirs = sorted((run_root / "profiles" / profile / "runs").glob("run_*")) + for run_dir in run_dirs: + meta = read_json(run_dir / "run-meta.json", {}) + if not meta: + continue + rows.append({ + "profile": profile, + "rp": "rpki-prover", + "run_seq": meta.get("runSeq", ""), + "sync_mode": meta.get("syncMode", ""), + "start_utc": meta.get("startUtc", ""), + "end_utc": meta.get("endUtc", ""), + "exit_code": meta.get("exitCode", ""), + "wall_ms": meta.get("timing", {}).get("wallMs", ""), + "max_rss_kb": meta.get("timing", {}).get("maxRssKb", ""), + "max_process_rss_kb": meta.get("timing", {}).get("maxProcessRssKb", ""), + "vrps": meta.get("counts", {}).get("vrpsRaw", ""), + "vrps_unique": meta.get("counts", {}).get("vrpsUnique", ""), + "vaps": "n/a", + "vaps_available": False, + "validation_algorithm": meta.get("validationAlgorithm", ""), + "remote_run_dir": str(run_dir), + }) + reports = run_root / "reports" + reports.mkdir(parents=True, exist_ok=True) + if rows: + with (reports / "per_run_metrics.csv").open("w", encoding="utf-8", newline="") as handle: + writer = csv.DictWriter(handle, fieldnames=list(rows[0])) + writer.writeheader() + writer.writerows(rows) + write_json(reports / "per_run_metrics.json", {"generatedAtUtc": utc_now(), "rows": rows}) + + def normalize_one(run_root: Path, profile: str, run_seq: int) -> None: script = ctl_root() / "scripts" / "normalize_and_collect.py" if not script.exists(): @@ -460,6 +816,18 @@ def parse_entry_rirs(entry: dict[str, Any]) -> list[str]: return parse_rirs(str(value)) +def write_experiment_versions(run_root: Path) -> None: + versions = read_json(ctl_root() / "versions.json", {}) + image = load_config().get("rpkiProverImage") + if image: + versions["rpki-prover"] = image + profiles = versions.setdefault("profiles", []) + for profile in ["rpki-prover-incremental", "rpki-prover-full"]: + if profile not in profiles: + profiles.append(profile) + write_json(run_root / "versions.json", versions) + + def run_profile(profile: str, exp_id: str, runs: int, interval_secs: int, rirs: list[str], append: bool = False) -> int: ensure_dirs() rp = PROFILES[profile] @@ -467,7 +835,7 @@ def run_profile(profile: str, exp_id: str, runs: int, interval_secs: int, rirs: profile_root = run_root / "profiles" / profile (run_root / "logs").mkdir(parents=True, exist_ok=True) (run_root / "reports").mkdir(parents=True, exist_ok=True) - (run_root / "versions.json").write_text((ctl_root() / "versions.json").read_text(encoding="utf-8"), encoding="utf-8") + write_experiment_versions(run_root) now = utc_now() if append: run_info = normalize_experiment_metadata(read_json(run_root / "run-info.json", {})) @@ -504,6 +872,9 @@ def run_profile(profile: str, exp_id: str, runs: int, interval_secs: int, rirs: "experimentRoot": str(run_root), "stateRoot": str(profile_root / "state"), } + if rp == "rpki-prover": + run_info["validationAlgorithm"] = cache_profile_name(profile) + run_info["rpkiProverImage"] = load_config().get("rpkiProverImage", {}) write_json(run_root / "run-info.json", run_info) status = { "state": "running", @@ -535,7 +906,7 @@ def run_profile(profile: str, exp_id: str, runs: int, interval_secs: int, rirs: time.sleep(sleep_seconds) last_start = time.time() snapshot = run_seq == 1 - mode = "snapshot" if snapshot else "delta" + mode = "snapshot" if snapshot else ("warm" if rp == "rpki-prover" else "delta") run_dir = profile_root / "runs" / f"run_{run_seq:04d}" if run_dir.exists() and any(run_dir.iterdir()): raise RuntimeError(f"refusing to overwrite existing run directory: {run_dir}") @@ -550,13 +921,34 @@ def run_profile(profile: str, exp_id: str, runs: int, interval_secs: int, rirs: argv, env = build_rpki_client_args(run_dir, state, rirs) elif rp == "fort": argv, env = build_fort_args(run_dir, state, rirs) + elif rp == "rpki-prover": + prepare_rpki_prover_state(state, rirs) + container_suffix = hashlib.sha1(f"{run_root}:{run_seq}".encode("utf-8")).hexdigest()[:12] + container_name = f"rpctl-prover-{container_suffix}" + argv = build_rpki_prover_command(profile, run_dir, state, container_name) else: raise RuntimeError(rp) run_dir.mkdir(parents=True, exist_ok=True) (run_dir / "command.txt").write_text(command_string(argv) + "\n", encoding="utf-8") - code = timed_run(run_dir, argv, env=env) - normalize_one(run_root, profile, run_seq) - meta = parse_latest_meta(run_root, profile, run_seq) + if rp == "rpki-prover": + code, max_rss_kb, max_process_rss_kb, wall_ms, _ = timed_rpki_prover_run( + profile, run_dir, state, container_name + ) + meta = write_rpki_prover_run_metadata( + run_root, + profile, + run_seq, + mode, + wall_ms, + max_rss_kb, + max_process_rss_kb, + code, + ) + write_rpki_prover_group_report(run_root, profile) + else: + code = timed_run(run_dir, argv, env=env) + normalize_one(run_root, profile, run_seq) + meta = parse_latest_meta(run_root, profile, run_seq) update_current( state="running", lastRunSeq=run_seq, @@ -571,7 +963,10 @@ def run_profile(profile: str, exp_id: str, runs: int, interval_secs: int, rirs: if code != 0: exit_code = code break - write_group_report(run_root, interval_secs, completed_seq) + if rp == "rpki-prover": + write_rpki_prover_group_report(run_root, profile) + else: + write_group_report(run_root, interval_secs, completed_seq) final_state = "success" if exit_code == 0 else "failed" update_current(state=final_state, finishedAtUtc=utc_now(), exitCode=exit_code) run_info = normalize_experiment_metadata(read_json(run_root / "run-info.json", {})) @@ -645,6 +1040,10 @@ def cmd_paths(args: argparse.Namespace) -> None: print(f"runs={runs_dir()}") print(f"logs={logs_dir()}") print(f"bundle={cfg.get('bundle','')}") + image = cfg.get("rpkiProverImage", {}) + print(f"rpki_prover_image={image.get('tag', '')}") + print(f"rpki_prover_image_id={image.get('imageId', '')}") + print(f"rpki_prover_image_archive={image.get('archive', '')}") cur = current_live_status() if cur.get("experimentRoot"): print(f"current_experiment_root={cur['experimentRoot']}") @@ -668,6 +1067,8 @@ def cmd_start(args: argparse.Namespace) -> None: raise SystemExit("--runs must be >= 1") if not (ctl_root() / "versions.json").exists(): raise SystemExit("rpctl is not initialized; run ./rpctl init --bundle first") + if PROFILES[profile] == "rpki-prover": + require_rpki_prover_image() exp_id = args.exp_id or f"{utc_id()}_{profile}_{args.runs}runs_{interval_secs}s" runner_args = [ sys.executable, @@ -717,6 +1118,8 @@ def cmd_continue(args: argparse.Namespace) -> None: if entry["state"] != "success": raise SystemExit(f"can only continue a successful experiment: exp_id={args.exp_id} state={entry['state']}") profile = entry["profile"] + if PROFILES[profile] == "rpki-prover": + require_rpki_prover_image() rirs = parse_entry_rirs(entry) interval_secs = parse_interval(args.interval) runner_args = [ @@ -867,6 +1270,9 @@ def parse_command_summary(command: str) -> str: keys.append(f"--rsync-scope={tokens[idx + 1]}") elif token.startswith("--rsync-scope="): keys.append(token) + if RPKI_PROVER_IMAGE_TAG in token_set: + algorithm = "full-every-iteration" if "--no-incremental-validation" in token_set else "incremental" + keys.append(f"algorithm={algorithm}") return " ".join(keys) @@ -945,6 +1351,10 @@ def cache_profile_name(profile: str) -> str: return "pp+roa+crt" if profile == "ours-no-cache": return "off" + if profile == "rpki-prover-incremental": + return "incremental" + if profile == "rpki-prover-full": + return "full-every-iteration" return "n/a" @@ -971,6 +1381,8 @@ def profile_label(profile: str) -> str: "routinator-latest-release": "routinator", "rpki-client-latest-release": "rpki-client", "fort-latest-release": "fort", + "rpki-prover-incremental": "prover-inc", + "rpki-prover-full": "prover-full", } return labels.get(profile, profile) @@ -1043,6 +1455,9 @@ def infer_output_counts(profile: str, run_dir: Path) -> tuple[str, str]: return vrps, vaps if rp == "fort": return count_csv_data_rows(run_dir / "vrps.csv"), "0" if (run_dir / "vrps.csv").exists() else "" + if rp == "rpki-prover": + raw, _ = collect_rpki_prover_vrps(run_dir / "vrps.csv") + return str(raw) if (run_dir / "vrps.csv").exists() else "", "n/a" if (run_dir / "vrps.csv").exists() else "" return "", "" @@ -1355,6 +1770,7 @@ def fallback_run_details(entry: dict[str, Any]) -> list[dict[str, Any]]: "wall_ms": timing.get("wallMs", ""), "max_rss_kb": timing.get("maxRssKb", ""), "vrps": first_non_empty(meta.get("counts", {}).get("vrps") if meta else "", inferred_vrps), + "vrps_unique": first_non_empty(meta.get("counts", {}).get("vrpsUnique") if meta else ""), "vaps": first_non_empty(meta.get("counts", {}).get("vaps") if meta else "", inferred_vaps), "remote_run_dir": str(run_dir), }) @@ -1391,7 +1807,7 @@ def cmd_view(args: argparse.Namespace) -> None: print("command=" + entry["command"]) if rows: print("\nper-run:") - headers = ["run", "mode", "exit", "start", "end", "wall_ms", "rss_kb", "vrps", "vaps", "path"] + headers = ["run", "mode", "exit", "start", "end", "wall_ms", "rss_kb", "vrps", "vrps_unique", "vaps", "path"] table_rows = [] for row in rows: table_rows.append([ @@ -1403,6 +1819,7 @@ def cmd_view(args: argparse.Namespace) -> None: first_non_empty(row.get("wall_ms"), row.get("wallMs")), first_non_empty(row.get("max_rss_kb"), row.get("maxRssKb")), first_non_empty(row.get("vrps"), row.get("VRPs")), + first_non_empty(row.get("vrps_unique"), row.get("vrpsUnique")), first_non_empty(row.get("vaps"), row.get("VAPs")), first_non_empty(row.get("remote_run_dir"), row.get("remoteRunDir")), ]) @@ -1412,7 +1829,7 @@ def cmd_view(args: argparse.Namespace) -> None: def build_parser() -> argparse.ArgumentParser: formatter = argparse.RawDescriptionHelpFormatter parser = argparse.ArgumentParser( - description="Control one local four-RP performance run on this server.", + description="Control one local five-RP performance run on this server.", formatter_class=formatter, epilog=""" RP/profile naming: @@ -1435,6 +1852,14 @@ RP/profile naming: fort Alias of fort-latest-release. Uses archived FORT validator release binary. + rpki-prover / rpki-prover-incremental + Uses the archived rpki-prover Docker image with its default incremental + validation algorithm and persistent state. + + rpki-prover-full / rpki-prover-no-incremental + Uses the same image with --no-incremental-validation. Its state is kept + separate from the incremental profile. + Cache difference for ours RP: --rp ours adds: --enable-publication-point-validation-cache @@ -1444,10 +1869,12 @@ Cache difference for ours RP: --rp ours-no-cache does not add those flags. Typical usage: - ./rpctl init --bundle /root/rpki_4rp_ctl_runtime_bundle + ./rpctl init --bundle /data/rpki_4rp_ctl_runtime_bundle \\ + --rpki-prover-image /data/rpki-prover-6857d4bf-linux-amd64.docker.tar.gz ./rpctl list ./rpctl start --rp ours --interval 10m --runs 6 --rirs all5 ./rpctl start --rp ours-no-cache --interval 10m --runs 6 --rirs all5 + ./rpctl start --rp rpki-prover --interval 0 --runs 2 --rirs all5 ./rpctl status ./rpctl show ./rpctl show --failed @@ -1458,7 +1885,8 @@ Typical usage: Data model: - Experiment data stays under ./runs///. - - Each experiment starts with a snapshot; later child runs are deltas and reuse the profile state. + - Each experiment starts with a snapshot; later child runs reuse the profile state. + - rpki-prover labels reused-state runs as warm because its native storage decides snapshot/delta behavior. - By default only one RP/profile can run at a time. """, ) @@ -1481,14 +1909,24 @@ The bundle must contain: scripts/write_group_report.py versions.json +Optional rpki-prover image archive: + Pass --rpki-prover-image to copy and load the saved linux/amd64 Docker image. + Init verifies image architecture, tag and rpki-prover version without pulling online. + Recommended source on 47.77.237.41: /root/rpki_4rp_ctl_runtime_bundle Example: - ./rpctl init --bundle /root/rpki_4rp_ctl_runtime_bundle + ./rpctl init --bundle /data/rpki_4rp_ctl_runtime_bundle \\ + --rpki-prover-image /data/rpki-prover-6857d4bf-linux-amd64.docker.tar.gz """, ) p.add_argument("--bundle", required=True, help="Path to the archived #100 runtime bundle root.") + p.add_argument( + "--rpki-prover-image", + default="", + help="Optional saved rpki-prover linux/amd64 Docker image archive. Existing four-RP profiles do not require it.", + ) p.set_defaults(func=cmd_init) p = sub.add_parser( @@ -1502,6 +1940,10 @@ Use this command when unsure which value to pass to --rp. Important ours RP profiles: ours-pp-object-cache cache-enabled profile ours-no-cache no-cache baseline profile + +rpki-prover profiles: + rpki-prover-incremental default incremental validation + rpki-prover-full --no-incremental-validation baseline """, ) p.set_defaults(func=cmd_list) @@ -1525,6 +1967,8 @@ RP/profile examples: ./rpctl start --rp routinator --interval 15m --runs 2 --rirs apnic ./rpctl start --rp rpki-client --interval 0 --runs 1 --rirs apnic,arin ./rpctl start --rp fort --interval 0 --runs 1 --rirs all5 + ./rpctl start --rp rpki-prover --interval 0 --runs 2 --rirs all5 + ./rpctl start --rp rpki-prover-full --interval 0 --runs 1 --rirs all5 Cache mapping: --rp ours enables PP cache + ROA cache + child certificate cache. @@ -1537,7 +1981,7 @@ Safety: p.add_argument( "--rp", required=True, - help="RP/profile or alias. Common values: ours, ours-no-cache, routinator, rpki-client, fort.", + help="RP/profile or alias. Common values: ours, ours-no-cache, routinator, rpki-client, fort, rpki-prover, rpki-prover-full.", ) p.add_argument( "--interval", @@ -1578,6 +2022,7 @@ Run semantics: Examples: ./rpctl continue --exp-id 20260709T081122Z_ours-pp-object-cache_2runs_600s --runs 3 --interval 10m ./rpctl continue --rp routinator --exp-id m18_237_routinator_all5_snapshot_delta_20260710T002726Z --runs 1 --interval 0 + ./rpctl continue --rp rpki-prover --exp-id prover_all5 --runs 1 --interval 0 """, ) p.add_argument("--exp-id", required=True, help="Existing successful experiment id to append to.") @@ -1692,7 +2137,7 @@ Examples: ./rpctl show --json """, ) - p.add_argument("--rp", default="", help="Optional RP/profile filter, e.g. ours, ours-no-cache, routinator, rpki-client, fort.") + p.add_argument("--rp", default="", help="Optional RP/profile filter, e.g. ours, ours-no-cache, routinator, rpki-client, fort, rpki-prover.") p.add_argument("--failed", action="store_true", help="Only show failed, partial, incomplete or exited experiments.") p.add_argument("--limit", type=int, default=50, help="Maximum rows to show. 0 means no limit. Default: 50.") p.add_argument("--paths", action="store_true", help="Also show full experiment root paths. Hidden by default to keep the table readable.") diff --git a/scripts/rpctl/test_rpctl.py b/scripts/rpctl/test_rpctl.py new file mode 100644 index 0000000..d1b44c9 --- /dev/null +++ b/scripts/rpctl/test_rpctl.py @@ -0,0 +1,113 @@ +from __future__ import annotations + +import importlib.machinery +import importlib.util +import json +import tempfile +import unittest +from pathlib import Path + + +def load_rpctl(): + path = Path(__file__).with_name("rpctl") + loader = importlib.machinery.SourceFileLoader("rpctl_module", str(path)) + spec = importlib.util.spec_from_loader(loader.name, loader) + module = importlib.util.module_from_spec(spec) + loader.exec_module(module) + return module + + +class RpkiProverProfileTests(unittest.TestCase): + @classmethod + def setUpClass(cls) -> None: + cls.rpctl = load_rpctl() + + def test_profile_aliases_and_algorithms(self) -> None: + self.assertEqual(self.rpctl.normalize_profile("rpki-prover"), "rpki-prover-incremental") + self.assertEqual(self.rpctl.normalize_profile("rpki-prover-no-incremental"), "rpki-prover-full") + self.assertNotIn("--no-incremental-validation", self.rpctl.rpki_prover_arguments("rpki-prover-incremental")) + self.assertIn("--no-incremental-validation", self.rpctl.rpki_prover_arguments("rpki-prover-full")) + incremental = self.rpctl.parse_command_summary( + f"docker run {self.rpctl.RPKI_PROVER_IMAGE_TAG} --once" + ) + full = self.rpctl.parse_command_summary( + f"docker run {self.rpctl.RPKI_PROVER_IMAGE_TAG} --once --no-incremental-validation" + ) + self.assertEqual(incremental, "algorithm=incremental") + self.assertEqual(full, "algorithm=full-every-iteration") + + def test_vrp_count_preserves_raw_and_unique(self) -> None: + with tempfile.TemporaryDirectory() as temp: + csv_path = Path(temp) / "vrps.csv" + csv_path.write_text( + "ASN,IP Prefix,Max Length,Trust Anchor\n" + "AS64496,192.0.2.0/24,24,TA-A\n" + "AS64496,192.0.2.0/24,24,TA-B\n" + "AS64497,2001:db8::/32,48,TA-A\n", + encoding="utf-8", + ) + raw, unique = self.rpctl.collect_rpki_prover_vrps(csv_path) + self.assertEqual(raw, 3) + self.assertEqual(len(unique), 2) + + def test_run_metadata_marks_vaps_not_applicable(self) -> None: + with tempfile.TemporaryDirectory() as temp: + root = Path(temp) + old_ctl_root = self.rpctl.ctl_root + self.rpctl.ctl_root = lambda: root + try: + self.rpctl.save_config({"rpkiProverImage": {"tag": self.rpctl.RPKI_PROVER_IMAGE_TAG}}) + run_root = root / "runs" / "rpki-prover-incremental" / "test" + run_dir = run_root / "profiles" / "rpki-prover-incremental" / "runs" / "run_0001" + run_dir.mkdir(parents=True) + (run_dir / "start-utc.txt").write_text("2026-07-17T00:00:00Z\n", encoding="utf-8") + (run_dir / "end-utc.txt").write_text("2026-07-17T00:01:00Z\n", encoding="utf-8") + (run_dir / "vrps.csv").write_text( + "ASN,IP Prefix,Max Length,Trust Anchor\nAS64496,192.0.2.0/24,24,TA-A\n", + encoding="utf-8", + ) + meta = self.rpctl.write_rpki_prover_run_metadata( + run_root, "rpki-prover-incremental", 1, "snapshot", 60_000, 1234, 1200, 0 + ) + self.assertEqual(meta["counts"]["vrpsRaw"], 1) + self.assertEqual(meta["counts"]["vrpsUnique"], 1) + self.assertEqual(meta["counts"]["vaps"], "n/a") + self.assertFalse(meta["vapsAvailable"]) + self.rpctl.write_rpki_prover_group_report(run_root, "rpki-prover-incremental") + report = (run_root / "reports" / "per_run_metrics.csv").read_text(encoding="utf-8") + self.assertIn("vrps_unique", report) + self.assertIn("n/a", report) + finally: + self.rpctl.ctl_root = old_ctl_root + + def test_missing_image_configuration_fails_before_docker(self) -> None: + with tempfile.TemporaryDirectory() as temp: + root = Path(temp) + old_ctl_root = self.rpctl.ctl_root + self.rpctl.ctl_root = lambda: root + try: + self.rpctl.save_config({}) + with self.assertRaisesRegex(SystemExit, "not initialized"): + self.rpctl.require_rpki_prover_image() + finally: + self.rpctl.ctl_root = old_ctl_root + + def test_versions_include_prover_profiles(self) -> None: + with tempfile.TemporaryDirectory() as temp: + root = Path(temp) + old_ctl_root = self.rpctl.ctl_root + self.rpctl.ctl_root = lambda: root + try: + (root / "versions.json").write_text(json.dumps({"profiles": ["fort-latest-release"]}), encoding="utf-8") + self.rpctl.save_config({"rpkiProverImage": {"tag": self.rpctl.RPKI_PROVER_IMAGE_TAG}}) + run_root = root / "experiment" + self.rpctl.write_experiment_versions(run_root) + versions = json.loads((run_root / "versions.json").read_text(encoding="utf-8")) + self.assertIn("rpki-prover-incremental", versions["profiles"]) + self.assertIn("rpki-prover-full", versions["profiles"]) + finally: + self.rpctl.ctl_root = old_ctl_root + + +if __name__ == "__main__": + unittest.main()