From 32c9cb5f9632dfb23db29a666e9113369f5b3bdb Mon Sep 17 00:00:00 2001 From: lucarlig Date: Wed, 16 Sep 2026 09:09:26 +0100 Subject: [PATCH] Add repeatable FYRE scaling benchmarks Signed-off-by: lucarlig --- CHANGELOG.md | 7 + Cargo.toml | 3 + README.md | 20 +- benchmarks/fyre/README.md | 85 ++ benchmarks/fyre/campaign.py | 792 +++++++++++++++++ benchmarks/fyre/deploy/dataplane.compose.yaml | 59 ++ benchmarks/fyre/deploy/fast-time.compose.yaml | 19 + benchmarks/fyre/deploy/monitor.py | 168 ++++ benchmarks/fyre/deploy/run_locust.py | 190 ++++ benchmarks/fyre/deploy/smoke.py | 80 ++ benchmarks/fyre/report.py | 182 ++++ benchmarks/fyre/scaling.yaml | 47 + benchmarks/fyre/terraform/.terraform.lock.hcl | 15 + benchmarks/fyre/terraform/main.tf | 63 ++ benchmarks/fyre/terraform/outputs.tf | 37 + benchmarks/fyre/terraform/variables.tf | 31 + benchmarks/fyre/terraform/versions.tf | 11 + benchmarks/fyre/test_campaign.py | 289 ++++++ docker/helpers.Dockerfile | 1 + scripts/locustfile_mcp.py | 143 ++- src/app.rs | 123 ++- src/app_tests.rs | 72 +- src/cli.rs | 45 + src/cli_public_tests.rs | 38 +- src/infrastructure/assets.rs | 14 + src/runtime/fyre.rs | 826 ++++++++++++++++++ src/runtime/mod.rs | 2 + 27 files changed, 3314 insertions(+), 48 deletions(-) create mode 100644 benchmarks/fyre/README.md create mode 100644 benchmarks/fyre/campaign.py create mode 100644 benchmarks/fyre/deploy/dataplane.compose.yaml create mode 100644 benchmarks/fyre/deploy/fast-time.compose.yaml create mode 100644 benchmarks/fyre/deploy/monitor.py create mode 100644 benchmarks/fyre/deploy/run_locust.py create mode 100644 benchmarks/fyre/deploy/smoke.py create mode 100644 benchmarks/fyre/report.py create mode 100644 benchmarks/fyre/scaling.yaml create mode 100644 benchmarks/fyre/terraform/.terraform.lock.hcl create mode 100644 benchmarks/fyre/terraform/main.tf create mode 100644 benchmarks/fyre/terraform/outputs.tf create mode 100644 benchmarks/fyre/terraform/variables.tf create mode 100644 benchmarks/fyre/terraform/versions.tf create mode 100644 benchmarks/fyre/test_campaign.py create mode 100644 src/runtime/fyre.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 3765be8..f4309a9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,13 @@ and this project uses [Semantic Versioning](https://semver.org/spec/v2.0.0.html) ### Added +- Add `load fyre run|status|destroy` (with short aliases) and a packaged FYRE + Terraform campaign for matched vertical/horizontal Rust dataplane scaling. + The campaign pins provider and container versions, uses dedicated Locust and + Fast Time VMs, grows saturated helpers, captures host/container telemetry, + preserves raw reports before cleanup, and produces JSON, CSV, and a + Slack-ready comparison PNG. + - Add `-w/--workers` to distribute load across local Locust processes, `-i/--isolate-cpus` to split Docker CPUs between the target and load generator, and `-m/--builtin-memory-limit` to tune the built-in gateway diff --git a/Cargo.toml b/Cargo.toml index d077aa7..68a69d2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -18,6 +18,9 @@ include = [ "/src/**", "/docker/**", "/scripts/locustfile_mcp.py", + "/benchmarks/fyre/**", + "!/benchmarks/fyre/**/__pycache__/**", + "!/benchmarks/fyre/**/*.pyc", "/scripts/live_protocol/sitecustomize.py", "/tests/conformance/baselines/**", "/README.md", diff --git a/README.md b/README.md index 6b7fcd9..f38ebcb 100644 --- a/README.md +++ b/README.md @@ -47,7 +47,7 @@ Every public command and option has a short form, shown in `--help`. | --- | --- | --- | | `stack` | `s` | `up` → `u`, `down` → `d`, `status` → `s`, `logs` → `l`, `config` → `c` | | `probe` | `p` | — | -| `load` | `l` | `run` → `r` | +| `load` | `l` | `run` → `r`, `fyre` → `f` (`run` → `r`, `status` → `s`, `destroy` → `d`) | | `live` | `v` | — | | `conformance` | `c` | `run` → `r`, `report` → `p` | | `debug` | `d` | `inspect` → `i`, `token` → `t` | @@ -219,6 +219,24 @@ routing snapshot in Redis, without the control plane. It discovers Fast Time's catalog directly; it never starts the conformance fixture or its proxy. Conformance, probes, and Inspector retain their protocol fixtures. +### FYRE scaling campaign + +Run the reproducible vertical and horizontal Rust dataplane comparison on FYRE: + +```bash +cf-integration load fyre run +cf-integration load fyre status --run-id scale-candidate +cf-integration load fyre destroy --run-id scale-candidate +``` + +The short forms are `cf-integration l f r`, `l f s`, and `l f d`; configuration +and run IDs use `-f` and `-i`. The packaged matrix, infrastructure lifecycle, +capacity-search rules, recovery behavior, and report layout are documented in +[`benchmarks/fyre/README.md`](benchmarks/fyre/README.md). FYRE credentials stay +in provider environment variables. All generated Terraform state, inventories, +raw reports, telemetry, manifests, and the Slack-ready PNG are kept under +`CF_INTEGRATION_DIR/fyre//`. + ## Live gateway checks Groups are `mcp`, `rbac`, `protocol`, and `all` (default): diff --git a/benchmarks/fyre/README.md b/benchmarks/fyre/README.md new file mode 100644 index 0000000..c6fa626 --- /dev/null +++ b/benchmarks/fyre/README.md @@ -0,0 +1,85 @@ +# FYRE dataplane scaling benchmark + +This benchmark compares vertical and horizontal Rust dataplane scaling with the +same total dataplane CPU and memory. It provisions a dedicated Locust VM, a +dedicated Fast Time VM, and one to three dataplane VMs. Each dataplane VM owns +its Redis and loopback JWKS helper, and receives the same routing snapshot and +ephemeral signing key. + +| Scenario | Dataplane allocation | Total allocation | +| --- | --- | --- | +| Baseline | 1 × 2 vCPU / 8 GB | 2 vCPU / 8 GB | +| Vertical 2× | 1 × 4 vCPU / 16 GB | 4 vCPU / 16 GB | +| Horizontal 2× | 2 × 2 vCPU / 8 GB | 4 vCPU / 16 GB | +| Vertical 3× | 1 × 6 vCPU / 24 GB | 6 vCPU / 24 GB | +| Horizontal 3× | 3 × 2 vCPU / 8 GB | 6 vCPU / 24 GB | +| Vertical 4× extension | 1 × 8 vCPU / 32 GB | 8 vCPU / 32 GB | + +## Prerequisites + +- Terraform 1.8+, or set `CF_TERRAFORM_BIN` to a compatible Terraform binary. +- `python3`, `uv`, SSH, and SCP on the orchestration host. +- An SSH key pair at the paths configured in `scaling.yaml`. +- FYRE provider credentials in `FYRE_USERNAME` and `FYRE_API_KEY`. +- Optionally set `FYRE_PRODUCT_GROUP_ID` and `FYRE_SITE`. Without a product + group the configuration uses quick-burn quota with an eight-hour TTL. + +Credential values are inherited by Terraform and are never copied into the +run manifest, command arguments, reports, or logs. + +## Run and recover + +```bash +cf-integration load fyre run +cf-integration l f r -f benchmarks/fyre/scaling.yaml -i scale-candidate + +cf-integration load fyre status --run-id scale-candidate +cf-integration l f s -i scale-candidate + +cf-integration load fyre destroy --run-id scale-candidate +cf-integration l f d -i scale-candidate +``` + +Generated state lives under +`$CF_INTEGRATION_DIR/fyre//`. The CLI copies Terraform into that +directory, so each run has isolated state. Resource names begin with the run ID +and `destroy` verifies the ownership file before using that state. Existing +manually created VMs are outside the state and cannot be deleted by the command. + +The run downloads each phase's Locust reports and host telemetry as it +finishes. It then builds `results/summary.json`, `results/summary.csv`, and +`results/slack-scaling.png` before destroying run-owned VMs. On error or +interrupt it retains already downloaded artifacts, retries Terraform cleanup +three times, and records `cleanup-failed` if manual `destroy` is needed. + +## Capacity method + +The workload uses modern MCP `2026-07-28`, `FastHttpUser`, multiple Locust +workers, and the six nonfailure Fast Time tools. Requests go directly to the +native endpoint of a dataplane replica; virtual users are assigned evenly +across replicas and the report retains per-replica request rates. + +Each concurrency step smokes every tool through every replica, ramps within +30 seconds, warms the backend for 30 seconds, and measures for 120 seconds. +The measured Locust phase resets statistics when spawning completes, and the +telemetry summary uses the same recorded measurement-window boundary. It starts +at 125 users and doubles until the first error or a two-step throughput plateau. +After an error it only tests lower concurrency while refining the boundary to +12.5 percent. The selected capacity must pass three measured repetitions with +zero request and worker errors. Each scenario is bounded at 32,000 users, and +the full provision-and-benchmark matrix stops after six hours before recovery +and cleanup. + +Locust and Fast Time start at 2 vCPU / 8 GB. Host and container telemetry +checks CPU, per-core use, memory, swap, pressure stalls, sockets, network +counters, worker exits, and virtualization steal. A saturated helper is grown +through the configured sizes. Any helper resize archives prior attempts under +`invalidated/` and restarts the matrix so final comparisons use the same helper +sizes. Reaching 16 vCPU / 32 GB without demonstrated headroom makes the +campaign inconclusive. + +The final report includes confirmed zero-error RPS, p50/p95/p99, vertical and +horizontal speedups, scaling efficiency, matched horizontal advantage, RPS per +allocated dataplane vCPU, repetition variability, resource inventory, CPU +model, and steal time. Redis and authentication helpers run on each dataplane +VM, so the result measures the complete dataplane deployment allocation. diff --git a/benchmarks/fyre/campaign.py b/benchmarks/fyre/campaign.py new file mode 100644 index 0000000..142babc --- /dev/null +++ b/benchmarks/fyre/campaign.py @@ -0,0 +1,792 @@ +"""Bootstrap FYRE hosts and find one scenario's zero-error capacity.""" + +from __future__ import annotations + +import argparse +import csv +import json +import shlex +import statistics +import subprocess +import tempfile +import time +from pathlib import Path + +HELPER_SATURATED = 42 + + +def run( + arguments: list[str], + *, + check: bool = True, + capture: bool = False, + timeout: float | None = None, +) -> subprocess.CompletedProcess[str]: + return subprocess.run( + arguments, + check=check, + text=True, + timeout=timeout, + stdout=subprocess.PIPE if capture else None, + stderr=subprocess.PIPE if capture else None, + ) + + +class Remote: + def __init__(self, user: str, key: Path, known_hosts: Path): + self.user = user + self.options = [ + "-i", + str(key), + "-o", + "BatchMode=yes", + "-o", + "IdentitiesOnly=yes", + "-o", + f"UserKnownHostsFile={known_hosts}", + "-o", + "StrictHostKeyChecking=accept-new", + "-o", + "ConnectTimeout=10", + ] + + def ssh( + self, + host: str, + command: str, + *, + check: bool = True, + capture: bool = False, + timeout: float | None = None, + ): + return run( + ["ssh", *self.options, f"{self.user}@{host}", command], + check=check, + capture=capture, + timeout=timeout, + ) + + def copy_to(self, host: str, source: Path, destination: str) -> None: + destination = destination.removeprefix("~/") + run(["scp", *self.options, str(source), f"{self.user}@{host}:{destination}"]) + + def copy_from( + self, + host: str, + source: str, + destination: Path, + *, + recursive: bool = False, + check: bool = True, + ) -> None: + if recursive: + destination.mkdir(parents=True, exist_ok=True) + else: + destination.parent.mkdir(parents=True, exist_ok=True) + source = source.removeprefix("~/") + arguments = ["scp", *self.options] + if recursive: + arguments.append("-r") + run([*arguments, f"{self.user}@{host}:{source}", str(destination)], check=check) + + +def wait_for_ssh(remote: Remote, host: str, deadline: float) -> None: + last = "not attempted" + while time.monotonic() < deadline: + result = remote.ssh(host, "true", check=False, capture=True, timeout=15) + if result.returncode == 0: + return + last = (result.stderr or result.stdout).strip()[-300:] + time.sleep(5) + raise RuntimeError(f"SSH host {host} was not ready: {last}") + + +def bootstrap(remote: Remote, host: str, deploy: Path) -> None: + wait_for_ssh(remote, host, time.monotonic() + 600) + remote.ssh( + host, + "if ! command -v docker >/dev/null || ! docker compose version >/dev/null 2>&1; then sudo DEBIAN_FRONTEND=noninteractive apt-get update -qq && sudo DEBIAN_FRONTEND=noninteractive apt-get install -y -qq docker.io docker-compose-v2 iproute2 && sudo usermod -aG docker $USER && sudo systemctl enable --now docker; fi; mkdir -p ~/cf-fyre/state/keys ~/cf-fyre/reports ~/cf-fyre/telemetry", + ) + for path in deploy.iterdir(): + if path.is_file(): + remote.copy_to(host, path, f"~/cf-fyre/{path.name}") + + +def write_remote_file( + remote: Remote, host: str, contents: str, destination: str, mode: int = 0o600 +) -> None: + destination = destination.removeprefix("~/") + with tempfile.NamedTemporaryFile("w", delete=False, encoding="utf-8") as stream: + stream.write(contents) + temporary = Path(stream.name) + try: + temporary.chmod(mode) + remote.copy_to(host, temporary, destination) + remote.ssh(host, f"chmod {mode:o} {shlex.quote(destination)}") + finally: + temporary.unlink(missing_ok=True) + + +def compose_up(remote: Remote, host: str, compose: str) -> None: + remote.ssh( + host, + f"cd ~/cf-fyre && docker compose --env-file benchmark.env -f {shlex.quote(compose)} pull && docker compose --env-file benchmark.env -f {shlex.quote(compose)} up -d --wait", + timeout=900, + ) + + +def prepare_hosts( + config: dict, inventory: dict, remote: Remote, deploy: Path, output: Path +) -> tuple[str, list[str]]: + hosts = [inventory["locust"], inventory["fast_time"], *inventory["dataplanes"]] + for host in hosts: + bootstrap(remote, host["public_ip"], deploy) + + images = config["images"] + fast_env = f"FAST_TIME_IMAGE={images['fast_time']}\n" + write_remote_file( + remote, inventory["fast_time"]["public_ip"], fast_env, "~/cf-fyre/benchmark.env" + ) + compose_up(remote, inventory["fast_time"]["public_ip"], "fast-time.compose.yaml") + + first = inventory["dataplanes"][0] + for index, target in enumerate(inventory["dataplanes"]): + allowed = ",".join( + [ + f"{target['private_ip']}:4445", + f"{target['public_ip']}:4445", + "127.0.0.1:4445", + "localhost:4445", + ] + ) + target_env = "\n".join( + [ + f"DATAPLANE_IMAGE={images['dataplane']}", + f"HELPERS_IMAGE={images['helpers']}", + f"REDIS_IMAGE={images['redis']}", + f"DATAPLANE_ALLOWED_HOSTS={allowed}", + f"CONFIG_CACHE_SECONDS={config['workload']['config_cache_seconds']}", + "", + ] + ) + write_remote_file( + remote, target["public_ip"], target_env, "~/cf-fyre/benchmark.env" + ) + if index > 0: + with tempfile.TemporaryDirectory() as temporary: + key = Path(temporary) / "jwt.key" + remote.copy_from( + first["public_ip"], "~/cf-fyre/state/keys/jwt.key", key + ) + remote.copy_to(target["public_ip"], key, "~/cf-fyre/state/keys/jwt.key") + remote.ssh( + target["public_ip"], "chmod 600 ~/cf-fyre/state/keys/jwt.key" + ) + compose_up(remote, target["public_ip"], "dataplane.compose.yaml") + if index == 0: + # The first auth container creates the campaign key; subsequent replicas receive it. + remote.ssh( + first["public_ip"], + "test -s ~/cf-fyre/state/keys/jwt.key && sudo chown $USER:$(id -gn) ~/cf-fyre/state/keys/jwt.key && chmod 600 ~/cf-fyre/state/keys/jwt.key", + ) + + token = remote.ssh( + first["public_ip"], + "cd ~/cf-fyre && docker compose --env-file benchmark.env -f dataplane.compose.yaml run --rm --no-deps config_writer token fyre-benchmark fyre-user", + capture=True, + ).stdout.strip() + if not token or "\n" in token: + raise RuntimeError("config helper did not return one bearer token") + token_file = output / ".token" + token_file.write_text(token, encoding="utf-8") + token_file.chmod(0o600) + try: + for target in inventory["dataplanes"]: + remote.copy_to(target["public_ip"], token_file, "~/cf-fyre/state/token") + remote.ssh(target["public_ip"], "chmod 600 ~/cf-fyre/state/token") + remote.ssh( + target["public_ip"], + 'cd ~/cf-fyre && export MCP_CONFORMANCE_TOKEN="$(cat state/token)" && docker compose --env-file benchmark.env -f dataplane.compose.yaml run --rm --no-deps -e MCP_CONFORMANCE_TOKEN config_writer fixture fyre-fast-time http://' + + inventory["fast_time"]["private_ip"] + + ":9080/mcp 2026-07-28", + timeout=120, + ) + locust_env = "\n".join( + [ + f"MCPGATEWAY_BEARER_TOKEN={token}", + "MCP_PROTOCOL_VERSION=2026-07-28", + "MCP_STACK_MODE=dataplane", + "MCP_SERVER_ID=fyre-fast-time", + "MCP_DIRECT_DATAPLANE=true", + "MCP_SKIP_TOOL_LIST=true", + "MCP_EXPLICIT_ZERO_DELAY=true", + "MCP_FYRE_WORKLOAD=true", + "MCP_TOOL_NAMES=convert_time,echo,get_stats,get_system_time,schema_success,verify-protocol", + "LOCUST_REQUEST_TIMEOUT_SECONDS=30", + "MCP_BASE_URLS=" + + ",".join( + f"http://{target['private_ip']}:4445" + for target in inventory["dataplanes"] + ), + "", + ] + ) + write_remote_file( + remote, + inventory["locust"]["public_ip"], + locust_env, + "~/cf-fyre/benchmark.secret.env", + ) + direct_env = "\n".join( + [ + f"MCPGATEWAY_BEARER_TOKEN={token}", + "MCP_PROTOCOL_VERSION=2026-07-28", + "MCP_STACK_MODE=controlplane", + "MCP_FYRE_WORKLOAD=true", + "MCP_SKIP_TOOL_LIST=true", + "MCP_EXPLICIT_ZERO_DELAY=true", + "MCP_TOOL_NAMES=convert_time,echo,get_stats,get_system_time,schema_success,verify-protocol", + "LOCUST_REQUEST_TIMEOUT_SECONDS=30", + f"MCP_BASE_URLS=http://{inventory['fast_time']['private_ip']}:9080", + "", + ] + ) + write_remote_file( + remote, + inventory["locust"]["public_ip"], + direct_env, + "~/cf-fyre/direct.secret.env", + ) + remote.copy_to( + inventory["locust"]["public_ip"], token_file, "~/cf-fyre/state/token" + ) + remote.ssh(inventory["locust"]["public_ip"], "chmod 600 ~/cf-fyre/state/token") + finally: + token_file.unlink(missing_ok=True) + urls = [ + f"http://{target['private_ip']}:4445/contextforge-rs/servers/fyre-fast-time/mcp" + for target in inventory["dataplanes"] + ] + return token, urls + + +def start_monitor(remote: Remote, host: str, role: str, name: str) -> int: + command = f"cd ~/cf-fyre && nohup python3 monitor.py --role {shlex.quote(role)} --output telemetry/{shlex.quote(name)}.jsonl >telemetry/{shlex.quote(name)}.log 2>&1 & echo $!" + return int(remote.ssh(host, command, capture=True).stdout.strip()) + + +def stop_monitor(remote: Remote, host: str, pid: int) -> None: + remote.ssh( + host, + f"kill -TERM {pid} 2>/dev/null || true; wait {pid} 2>/dev/null || true", + check=False, + ) + + +def smoke(remote: Remote, locust: dict, urls: list[str], locust_image: str) -> None: + command = " ".join( + [ + "cd ~/cf-fyre && docker run --rm --network host --entrypoint python", + "-v $HOME/cf-fyre:/work -w /work", + shlex.quote(locust_image), + "python smoke.py --urls", + shlex.quote(",".join(urls)), + "--token-file state/token", + ] + ) + remote.ssh(locust["public_ip"], command, timeout=120) + + +def read_stats(path: Path, use_aggregate: bool = False) -> dict: + with path.open(newline="", encoding="utf-8") as stream: + rows = list(csv.DictReader(stream)) + tool_rows = [ + row for row in rows if row.get("Name", "").startswith("MCP tools/call") + ] + if not tool_rows: + raise RuntimeError(f"Locust report {path} contains no measured tool traffic") + aggregate = next((row for row in rows if row.get("Name") == "Aggregated"), None) + selected = [aggregate] if use_aggregate and aggregate is not None else tool_rows + failures = sum(int(float(row.get("Failure Count") or 0)) for row in selected) + requests = sum(int(float(row.get("Request Count") or 0)) for row in selected) + rps = sum(float(row.get("Requests/s") or 0) for row in selected) + per_replica = {row["Name"]: float(row.get("Requests/s") or 0) for row in tool_rows} + + def weighted(column: str) -> float: + return ( + sum( + float(row.get(column) or 0) * int(float(row.get("Request Count") or 0)) + for row in selected + ) + / requests + if requests + else 0.0 + ) + + return { + "requests": requests, + "failures": failures, + "rps": rps, + "p50_ms": weighted("50%"), + "p95_ms": weighted("95%"), + "p99_ms": weighted("99%"), + "per_replica_rps": per_replica, + } + + +def kernel_counter(text: str, name: str) -> int: + lines = text.splitlines() + for header, values in zip(lines[0::2], lines[1::2]): + header_fields = header.split() + value_fields = values.split() + if not header_fields or not value_fields or header_fields[0] != value_fields[0]: + continue + try: + return int(value_fields[header_fields.index(name)]) + except (ValueError, IndexError): + continue + return 0 + + +def docker_pressure(text: str) -> bool: + for line in text.splitlines(): + try: + state = json.loads(line) + except ValueError: + continue + health = state.get("Health") or {} + if ( + state.get("OOMKilled") is True + or state.get("Status") == "dead" + or (state.get("Status") == "exited" and state.get("ExitCode") != 0) + or health.get("Status") == "unhealthy" + ): + return True + return False + + +def pressure(path: Path, after: float | None = None) -> dict[str, float | bool]: + samples = [] + for line in path.read_text(encoding="utf-8").splitlines(): + item = json.loads(line) + if item.get("kind") == "sample" and ( + after is None or item.get("time", 0) >= after + ): + samples.append(item) + busy = [ + item["cpu"]["cpu"]["busy_percent"] + for item in samples + if "cpu" in item.get("cpu", {}) + ] + memory = [item["memory"]["used_percent"] for item in samples] + core_names = { + key for item in samples for key in item.get("cpu", {}) if key != "cpu" + } + core_means = [ + statistics.fmean( + item["cpu"][key]["busy_percent"] + for item in samples + if key in item.get("cpu", {}) + ) + for key in core_names + ] + steal = [ + item["cpu"]["cpu"]["steal_percent"] + for item in samples + if "cpu" in item.get("cpu", {}) + ] + network_counters = [ + sum( + kernel_counter(item.get("netstat", ""), counter) + for counter in ("ListenOverflows", "ListenDrops", "TCPBacklogDrop") + ) + for item in samples + ] + docker_unhealthy = any( + docker_pressure(item.get("docker_state", "")) for item in samples + ) + return { + "mean_cpu_percent": statistics.fmean(busy) if busy else 0.0, + "max_memory_percent": max(memory, default=0.0), + "max_mean_core_percent": max(core_means, default=0.0), + "mean_steal_percent": statistics.fmean(steal) if steal else 0.0, + "worker_or_network_pressure": docker_unhealthy + or (len(network_counters) > 1 and network_counters[-1] > network_counters[0]), + } + + +def one_phase( + remote: Remote, + config: dict, + inventory: dict, + urls: list[str], + output: Path, + users: int, + seconds: int, + label: str, + env_file: str = "benchmark.secret.env", + measurement: bool = False, +) -> dict: + locust = inventory["locust"] + workers = max(2, int(config["active_helper"]["locust_cpu"]) - 1) + spawn_rate = max(1.0, users / config["workload"]["ramp_seconds"]) + total_seconds = seconds + config["workload"]["ramp_seconds"] + remote_output = f"reports/{label}" + monitors: list[tuple[str, int]] = [] + monitor_hosts = [ + (locust, "locust"), + (inventory["fast_time"], "fast-time"), + *[ + (target, f"dataplane-{index + 1}") + for index, target in enumerate(inventory["dataplanes"]) + ], + ] + for host, role in monitor_hosts: + monitors.append( + (host["public_ip"], start_monitor(remote, host["public_ip"], role, label)) + ) + try: + command = " ".join( + [ + "cd ~/cf-fyre && python3 run_locust.py", + "--image", + shlex.quote(config["images"]["locust"]), + "--users", + str(users), + "--spawn-rate", + str(spawn_rate), + "--seconds", + str(total_seconds), + "--workers", + str(workers), + "--output", + shlex.quote(remote_output), + "--env-file", + shlex.quote(env_file), + ] + ) + if measurement: + command += f" --reset-stats --measurement-seconds {seconds}" + result = remote.ssh( + locust["public_ip"], command, check=False, timeout=total_seconds + 180 + ) + finally: + for host, pid in monitors: + stop_monitor(remote, host, pid) + local = output / label + local.mkdir(parents=True, exist_ok=True) + remote.copy_from( + locust["public_ip"], + f"~/cf-fyre/{remote_output}/.", + local, + recursive=True, + check=False, + ) + pressures = {} + measurement_start = None + marker = local / "measurement-start.txt" + if measurement: + if not marker.is_file(): + return { + "passed": False, + "reason": "Locust did not record the measurement-window start", + "pressure": pressures, + } + measurement_start = float(marker.read_text(encoding="utf-8").strip()) + for host, role in monitor_hosts: + path = local / f"{role}.jsonl" + remote.copy_from( + host["public_ip"], f"~/cf-fyre/telemetry/{label}.jsonl", path, check=False + ) + if path.exists(): + pressures[role] = pressure(path, after=measurement_start) + if result.returncode != 0: + return { + "passed": False, + "reason": f"Locust exited {result.returncode}", + "pressure": pressures, + } + stats = read_stats(local / "locust_stats.csv", use_aggregate=measurement) + stats.update( + {"passed": stats["failures"] == 0, "pressure": pressures, "users": users} + ) + return stats + + +def helper_saturation(config: dict, result: dict) -> str | None: + workload = config["workload"] + for role in ("locust", "fast-time"): + values = result.get("pressure", {}).get(role, {}) + if values.get("mean_cpu_percent", 0) > workload["helper_cpu_percent"]: + return role + if values.get("max_memory_percent", 0) > workload["helper_memory_percent"]: + return role + if values.get("max_mean_core_percent", 0) > workload["worker_core_percent"]: + return role + if values.get("worker_or_network_pressure", False): + return role + return None + + +def measured_step( + remote: Remote, + config: dict, + inventory: dict, + urls: list[str], + output: Path, + users: int, + name: str, +) -> dict: + smoke(remote, inventory["locust"], urls, config["images"]["locust"]) + warmup = one_phase( + remote, + config, + inventory, + urls, + output, + users, + config["workload"]["warmup_seconds"], + f"{name}-warmup", + ) + if not warmup.get("passed"): + return warmup + result = one_phase( + remote, + config, + inventory, + urls, + output, + users, + config["workload"]["measure_seconds"], + name, + measurement=True, + ) + saturated = helper_saturation(config, result) + if saturated: + (output / "helper-request.json").write_text( + json.dumps({"role": saturated}, indent=2) + "\n", encoding="utf-8" + ) + raise SystemExit(HELPER_SATURATED) + return result + + +def capacity_search( + remote: Remote, config: dict, inventory: dict, urls: list[str], output: Path +) -> dict: + workload = config["workload"] + started = time.monotonic() + passing: list[dict] = [] + failing: dict | None = None + improvements: list[float] = [] + users = workload["first_users"] + step = 0 + while ( + users <= workload["maximum_users"] + and time.monotonic() - started < workload["maximum_campaign_seconds"] + ): + step += 1 + result = measured_step( + remote, config, inventory, urls, output, users, f"search-{step}-{users}" + ) + if not result.get("passed"): + failing = {"users": users, **result} + break + if passing: + improvements.append(100.0 * (result["rps"] / passing[-1]["rps"] - 1.0)) + passing.append(result) + if len(improvements) >= 2 and all( + value < workload["plateau_improvement_percent"] + for value in improvements[-2:] + ): + break + if users == workload["maximum_users"]: + break + users = min(workload["maximum_users"], users * 2) + + if not passing: + return { + "status": "failed", + "reason": "no zero-error concurrency passed", + "failing": failing, + } + if failing: + low = passing[-1]["users"] + high = failing["users"] + while (high - low) / high > workload["boundary_percent"] / 100.0: + users = (low + high) // 2 + result = measured_step( + remote, config, inventory, urls, output, users, f"refine-{users}" + ) + if result.get("passed"): + passing.append(result) + low = users + else: + failing = {"users": users, **result} + high = users + + candidate = max(passing, key=lambda item: item["users"]) + confirmations = [] + for repetition in range(workload["repetitions"]): + result = measured_step( + remote, + config, + inventory, + urls, + output, + candidate["users"], + f"confirm-{repetition + 1}-{candidate['users']}", + ) + if not result.get("passed"): + return { + "status": "failed-confirmation", + "candidate": candidate, + "confirmations": confirmations, + "failure": result, + } + confirmations.append(result) + direct_url = f"http://{inventory['fast_time']['private_ip']}:9080/mcp" + smoke(remote, inventory["locust"], [direct_url], config["images"]["locust"]) + direct_warmup = one_phase( + remote, + config, + inventory, + [direct_url], + output, + candidate["users"], + workload["warmup_seconds"], + "calibration-warmup", + "direct.secret.env", + ) + if not direct_warmup.get("passed"): + return { + "status": "inconclusive", + "reason": "direct Fast Time calibration warmup failed", + "calibration": direct_warmup, + } + calibration = one_phase( + remote, + config, + inventory, + [direct_url], + output, + candidate["users"], + workload["measure_seconds"], + "calibration", + "direct.secret.env", + measurement=True, + ) + saturated = helper_saturation(config, calibration) + if saturated: + (output / "helper-request.json").write_text( + json.dumps({"role": saturated}, indent=2) + "\n", encoding="utf-8" + ) + raise SystemExit(HELPER_SATURATED) + if ( + not calibration.get("passed") + or calibration.get("rps", 0) < min(item["rps"] for item in confirmations) * 1.05 + ): + return { + "status": "inconclusive", + "reason": "direct Fast Time calibration did not demonstrate five percent upstream headroom", + "calibration": calibration, + } + rps_values = [result["rps"] for result in confirmations] + imbalances = [] + for result in confirmations: + replicas = list(result["per_replica_rps"].values()) + mean = statistics.fmean(replicas) if replicas else 0.0 + imbalances.append( + 100.0 * (max(replicas) - min(replicas)) / mean + if mean and len(replicas) > 1 + else 0.0 + ) + best = min(confirmations, key=lambda item: item["rps"]) + return { + "status": "confirmed", + "users": candidate["users"], + "search": passing, + "failing": failing, + "confirmations": confirmations, + "rps": statistics.fmean(rps_values), + "rps_min": min(rps_values), + "rps_max": max(rps_values), + "rps_cv_percent": 100.0 + * statistics.pstdev(rps_values) + / statistics.fmean(rps_values) + if len(rps_values) > 1 + else 0.0, + "replica_imbalance_percent": statistics.fmean(imbalances), + "p50_ms": statistics.fmean(item["p50_ms"] for item in confirmations), + "p95_ms": statistics.fmean(item["p95_ms"] for item in confirmations), + "p99_ms": statistics.fmean(item["p99_ms"] for item in confirmations), + "lower_bound": candidate["users"] == workload["maximum_users"] + and failing is None, + "conservative_confirmation": best, + "direct_backend_calibration": calibration, + } + + +def collect_recovery(remote: Remote, inventory: dict, output: Path) -> None: + recovery = output / "recovery" + recovery.mkdir(parents=True, exist_ok=True) + for host in [inventory["locust"], inventory["fast_time"], *inventory["dataplanes"]]: + remote.ssh( + host["public_ip"], + "pkill -TERM -f 'python3 monitor.py' 2>/dev/null || true; docker ps --filter name=cf-fyre --format '{{.ID}}' | xargs -r docker rm -f >/dev/null 2>&1 || true", + check=False, + ) + destination = recovery / host["name"] + destination.mkdir(exist_ok=True) + remote.copy_from( + host["public_ip"], + "cf-fyre/reports/.", + destination / "reports", + recursive=True, + check=False, + ) + remote.copy_from( + host["public_ip"], + "cf-fyre/telemetry/.", + destination / "telemetry", + recursive=True, + check=False, + ) + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("--config", required=True) + parser.add_argument("--inventory", required=True) + parser.add_argument("--scenario", required=True) + parser.add_argument("--deploy", required=True) + parser.add_argument("--output", required=True) + parser.add_argument("--collect-only", action="store_true") + args = parser.parse_args() + config = json.loads(Path(args.config).read_text(encoding="utf-8")) + inventory = json.loads(Path(args.inventory).read_text(encoding="utf-8")) + scenario = next(item for item in config["scenarios"] if item["id"] == args.scenario) + output = Path(args.output) + output.mkdir(parents=True, exist_ok=True) + known_hosts = output.parent.parent / "known_hosts" + known_hosts.touch(exist_ok=True) + remote = Remote( + config["infrastructure"]["ssh_user"], + Path(config["resolved_ssh_private_key"]), + known_hosts, + ) + if args.collect_only: + collect_recovery(remote, inventory, output) + return + _, urls = prepare_hosts(config, inventory, remote, Path(args.deploy), output) + result = capacity_search(remote, config, inventory, urls, output) + result["scenario"] = scenario + result["inventory"] = inventory + (output / "result.json").write_text( + json.dumps(result, indent=2, sort_keys=True) + "\n", encoding="utf-8" + ) + if result["status"] != "confirmed": + raise SystemExit(1) + + +if __name__ == "__main__": + main() diff --git a/benchmarks/fyre/deploy/dataplane.compose.yaml b/benchmarks/fyre/deploy/dataplane.compose.yaml new file mode 100644 index 0000000..ec4c1f1 --- /dev/null +++ b/benchmarks/fyre/deploy/dataplane.compose.yaml @@ -0,0 +1,59 @@ +services: + redis: + image: ${REDIS_IMAGE:?Set REDIS_IMAGE to a pinned digest} + restart: unless-stopped + command: ["redis-server", "--save", "", "--appendonly", "no"] + healthcheck: + test: ["CMD", "redis-cli", "ping"] + interval: 2s + timeout: 2s + retries: 60 + + dataplane: + image: ${DATAPLANE_IMAGE:?Set DATAPLANE_IMAGE to a pinned digest} + restart: unless-stopped + ports: ["4445:4445"] + expose: ["4445"] + ulimits: + nofile: + soft: 65536 + hard: 65536 + environment: + CONTEXTFORGE_DATA_PLANE_ADDRESS: 0.0.0.0:4445 + CONTEXTFORGE_DATA_PLANE_REDIS_HOSTNAME: redis + CONTEXTFORGE_DATA_PLANE_REDIS_PORT: "6379" + CONTEXTFORGE_DATA_PLANE_REDIS_CONNECTION_MODE: plain-text + CONTEXTFORGE_DATA_PLANE_JWKS_URL: http://127.0.0.1:4446/.well-known/jwks.json + CONTEXTFORGE_DATA_PLANE_UPSTREAM_CONNECTION_MODE: plain-text-or-tls + CONTEXTFORGE_GATEWAY_RS_MCP_ALLOWED_HOSTS: ${DATAPLANE_ALLOWED_HOSTS:?Set DATAPLANE_ALLOWED_HOSTS} + CONTEXTFORGE_GATEWAY_RS_MCP_ALLOWED_ORIGINS: "" + CONTEXTFORGE_DATA_PLANE_USER_CONFIG_CACHE_EXPIRY_SECONDS: ${CONFIG_CACHE_SECONDS:-60} + RUST_LOG: warn + depends_on: + redis: + condition: service_healthy + + auth: + image: ${HELPERS_IMAGE:?Set HELPERS_IMAGE to a pinned digest} + restart: unless-stopped + network_mode: service:dataplane + volumes: ["./state/keys:/keys"] + command: ["__helper", "auth"] + healthcheck: + test: ["CMD", "cf-integration", "__helper", "health"] + interval: 2s + timeout: 2s + retries: 60 + depends_on: ["dataplane"] + + config_writer: + profiles: ["helpers"] + image: ${HELPERS_IMAGE:?Set HELPERS_IMAGE to a pinned digest} + network_mode: service:dataplane + volumes: ["./state/keys:/keys"] + environment: + CF_CONFIG_REDIS_URL: redis://redis:6379 + entrypoint: ["cf-integration", "__helper"] + depends_on: + redis: + condition: service_healthy diff --git a/benchmarks/fyre/deploy/fast-time.compose.yaml b/benchmarks/fyre/deploy/fast-time.compose.yaml new file mode 100644 index 0000000..db6a1f3 --- /dev/null +++ b/benchmarks/fyre/deploy/fast-time.compose.yaml @@ -0,0 +1,19 @@ +services: + fast_time: + image: ${FAST_TIME_IMAGE:?Set FAST_TIME_IMAGE to a pinned digest} + restart: unless-stopped + network_mode: host + command: [] + environment: + BIND_ADDRESS: 0.0.0.0:9080 + RUST_LOG: warn + ulimits: + nofile: + soft: 65536 + hard: 65536 + healthcheck: + test: ["CMD", "curl", "-fsS", "http://127.0.0.1:9080/health"] + interval: 2s + timeout: 2s + retries: 60 + start_period: 2s diff --git a/benchmarks/fyre/deploy/monitor.py b/benchmarks/fyre/deploy/monitor.py new file mode 100644 index 0000000..67ec236 --- /dev/null +++ b/benchmarks/fyre/deploy/monitor.py @@ -0,0 +1,168 @@ +"""Sample Linux host pressure as JSON lines without third-party packages.""" + +from __future__ import annotations + +import argparse +import json +import os +import platform +import socket +import subprocess +import time +from pathlib import Path + + +def read(path: str) -> str: + try: + return Path(path).read_text(encoding="utf-8", errors="replace") + except OSError as error: + return f"error={error}" + + +def cpu_times() -> dict[str, list[int]]: + rows: dict[str, list[int]] = {} + for line in read("/proc/stat").splitlines(): + fields = line.split() + if fields and fields[0].startswith("cpu"): + rows[fields[0]] = [int(value) for value in fields[1:]] + return rows + + +def cpu_percent( + previous: dict[str, list[int]], current: dict[str, list[int]] +) -> dict[str, dict[str, float]]: + result = {} + for name, values in current.items(): + before = previous.get(name, values) + deltas = [max(0, after - old) for old, after in zip(before, values)] + total = sum(deltas) or 1 + idle = sum(deltas[index] for index in (3, 4) if index < len(deltas)) + steal = deltas[7] if len(deltas) > 7 else 0 + result[name] = { + "busy_percent": round(100.0 * (total - idle) / total, 3), + "steal_percent": round(100.0 * steal / total, 3), + } + return result + + +def memory() -> dict[str, int | float]: + values = {} + for line in read("/proc/meminfo").splitlines(): + key, _, rest = line.partition(":") + try: + values[key] = int(rest.split()[0]) + except (IndexError, ValueError): + continue + total = values.get("MemTotal", 0) + available = values.get("MemAvailable", 0) + return { + "total_kib": total, + "available_kib": available, + "used_percent": round(100.0 * (total - available) / total, 3) if total else 0.0, + "swap_total_kib": values.get("SwapTotal", 0), + "swap_free_kib": values.get("SwapFree", 0), + } + + +def command_output(command: list[str]) -> str: + try: + return subprocess.run( + command, check=False, text=True, capture_output=True, timeout=3 + ).stdout.strip() + except (OSError, subprocess.TimeoutExpired) as error: + return f"error={error}" + + +def docker_state() -> str: + container_ids = [ + container_id + for container_id in command_output( + ["docker", "ps", "--all", "--quiet"] + ).splitlines() + if container_id and not container_id.startswith("error=") + ] + if not container_ids: + return "[]" + return command_output( + [ + "docker", + "inspect", + "--format", + "{{json .State}}", + *container_ids, + ] + ) + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("--output", required=True) + parser.add_argument("--role", required=True) + parser.add_argument("--interval", type=float, default=1.0) + args = parser.parse_args() + output = Path(args.output) + output.parent.mkdir(parents=True, exist_ok=True) + previous = cpu_times() + with output.open("a", encoding="utf-8", buffering=1) as stream: + stream.write( + json.dumps( + { + "kind": "host", + "role": args.role, + "hostname": socket.gethostname(), + "cpu_model": next( + ( + line.partition(":")[2].strip() + for line in read("/proc/cpuinfo").splitlines() + if line.startswith("model name") + ), + platform.processor(), + ), + "logical_cpus": os.cpu_count(), + "kernel": platform.release(), + }, + sort_keys=True, + ) + + "\n" + ) + while True: + time.sleep(args.interval) + current = cpu_times() + stream.write( + json.dumps( + { + "kind": "sample", + "time": time.time(), + "cpu": cpu_percent(previous, current), + "memory": memory(), + "loadavg": read("/proc/loadavg").strip(), + "pressure_cpu": read("/proc/pressure/cpu").strip(), + "pressure_memory": read("/proc/pressure/memory").strip(), + "vmstat": read("/proc/vmstat").strip(), + "sockstat": read("/proc/net/sockstat").strip(), + "netstat": read("/proc/net/netstat").strip(), + "snmp": read("/proc/net/snmp").strip(), + "network": read("/proc/net/dev").strip(), + "ss": command_output(["ss", "-s"]), + "processes": command_output( + [ + "ps", + "-eo", + "pid,ppid,comm,%cpu,%mem,rss,vsz,stat", + "--sort=-%cpu", + ] + ), + "docker": command_output( + ["docker", "stats", "--no-stream", "--format", "{{json .}}"] + ), + "docker_state": docker_state(), + }, + sort_keys=True, + ) + + "\n" + ) + previous = current + + +if __name__ == "__main__": + main() diff --git a/benchmarks/fyre/deploy/run_locust.py b/benchmarks/fyre/deploy/run_locust.py new file mode 100644 index 0000000..dbea77a --- /dev/null +++ b/benchmarks/fyre/deploy/run_locust.py @@ -0,0 +1,190 @@ +"""Run one distributed, headless Locust phase and propagate any worker failure.""" + +from __future__ import annotations + +import argparse +import os +import signal +import subprocess +import sys +import time +from pathlib import Path + +CONTAINERS: list[str] = [] + + +def docker( + *arguments: str, check: bool = True, capture: bool = False +) -> subprocess.CompletedProcess[str]: + return subprocess.run( + ["docker", *arguments], + check=check, + text=True, + stdout=subprocess.PIPE if capture else None, + stderr=subprocess.STDOUT if capture else None, + ) + + +def cleanup() -> None: + if CONTAINERS: + docker("rm", "--force", *CONTAINERS, check=False, capture=True) + + +def stop(_signal: int, _frame) -> None: + cleanup() + raise SystemExit(130) + + +def container_state(name: str) -> tuple[str, int]: + result = docker( + "inspect", + "--format", + "{{.State.Status}} {{.State.ExitCode}}", + name, + check=False, + capture=True, + ) + if result.returncode != 0: + return "missing", 1 + fields = result.stdout.strip().split() + if len(fields) != 2: + return "invalid", 1 + try: + return fields[0], int(fields[1]) + except ValueError: + return "invalid", 1 + + +def wait_for_cluster(master: str, workers: list[str]) -> int: + while True: + master_state, master_exit = container_state(master) + if master_state in {"exited", "dead", "missing", "invalid"}: + return master_exit + for worker in workers: + worker_state, _worker_exit = container_state(worker) + if worker_state != "running": + docker("stop", "--time", "1", master, check=False, capture=True) + return 1 + time.sleep(0.5) + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("--image", required=True) + parser.add_argument("--users", type=int, required=True) + parser.add_argument("--spawn-rate", type=float, required=True) + parser.add_argument("--seconds", type=int, required=True) + parser.add_argument("--workers", type=int, required=True) + parser.add_argument("--output", required=True) + parser.add_argument("--env-file", required=True) + parser.add_argument("--reset-stats", action="store_true") + parser.add_argument("--measurement-seconds", type=int) + args = parser.parse_args() + if min(args.users, args.spawn_rate, args.seconds, args.workers) <= 0: + parser.error("users, spawn-rate, seconds, and workers must be positive") + if args.measurement_seconds is not None and args.measurement_seconds <= 0: + parser.error("measurement-seconds must be positive") + if args.reset_stats != (args.measurement_seconds is not None): + parser.error("reset-stats and measurement-seconds must be used together") + + signal.signal(signal.SIGINT, stop) + signal.signal(signal.SIGTERM, stop) + output = Path(args.output) + output.mkdir(parents=True, exist_ok=True) + prefix = f"cf-fyre-{os.getpid()}" + master = f"{prefix}-master" + CONTAINERS.append(master) + common = [ + "--network", + "host", + "--ulimit", + "nofile=65536:65536", + "--env-file", + args.env_file, + "--volume", + f"{Path.cwd() / 'locustfile_mcp.py'}:/mnt/locust-cf/locustfile_mcp.py:ro", + "--volume", + f"{output.resolve()}:/mnt/reports", + ] + if args.reset_stats: + common.extend( + [ + "--env", + "MCP_MEASUREMENT_MARKER=/mnt/reports/measurement-start.txt", + "--env", + f"MCP_MEASUREMENT_SECONDS={args.measurement_seconds}", + ] + ) + run_seconds = ( + args.measurement_seconds + 60 + if args.measurement_seconds is not None + else args.seconds + ) + master_args = [ + "run", + "--detach", + "--name", + master, + *common, + args.image, + "-f", + "/mnt/locust-cf/locustfile_mcp.py", + "--master", + "--expect-workers", + str(args.workers), + "--headless", + "--users", + str(args.users), + "--spawn-rate", + str(args.spawn_rate), + "--run-time", + f"{run_seconds}s", + "--stop-timeout", + "1", + "--host", + "http://127.0.0.1", + "--csv", + "/mnt/reports/locust", + "--csv-full-history", + "--html", + "/mnt/reports/locust.html", + "--json-file", + "/mnt/reports/locust.json", + "--logfile", + "/mnt/reports/locust.log", + ] + if args.reset_stats: + master_args.append("--reset-stats") + docker(*master_args) + try: + workers = [] + for index in range(args.workers): + name = f"{prefix}-worker-{index + 1}" + CONTAINERS.append(name) + workers.append(name) + docker( + "run", + "--detach", + "--name", + name, + *common, + args.image, + "-f", + "/mnt/locust-cf/locustfile_mcp.py", + "--worker", + "--master-host", + "127.0.0.1", + ) + status = wait_for_cluster(master, workers) + for name in CONTAINERS: + state, exit_code = container_state(name) + if state in {"exited", "dead"} and exit_code != 0: + status = 1 + sys.exit(status) + finally: + time.sleep(0.2) + cleanup() + + +if __name__ == "__main__": + main() diff --git a/benchmarks/fyre/deploy/smoke.py b/benchmarks/fyre/deploy/smoke.py new file mode 100644 index 0000000..c0083bb --- /dev/null +++ b/benchmarks/fyre/deploy/smoke.py @@ -0,0 +1,80 @@ +"""Call every measured Fast Time tool through every dataplane replica.""" + +from __future__ import annotations + +import argparse +import json +import urllib.request +import uuid + +TOOLS = { + "convert_time": { + "time": "12:00", + "source_timezone": "UTC", + "target_timezone": "Europe/Dublin", + }, + "echo": {"message": "cf-integration", "delay": 0}, + "get_stats": {}, + "get_system_time": {"timezone": "UTC"}, + "schema_success": {}, + "verify-protocol": {}, +} + + +def call(url: str, token: str, tool: str, arguments: dict) -> None: + payload = { + "jsonrpc": "2.0", + "id": str(uuid.uuid4()), + "method": "tools/call", + "params": { + "name": tool, + "arguments": arguments, + "_meta": { + "io.modelcontextprotocol/protocolVersion": "2026-07-28", + "io.modelcontextprotocol/clientInfo": { + "name": "cf-integration-smoke", + "version": "1.0", + }, + "io.modelcontextprotocol/clientCapabilities": {}, + }, + }, + } + request = urllib.request.Request( + url, + data=json.dumps(payload).encode(), + headers={ + "Accept": "application/json, text/event-stream", + "Authorization": f"Bearer {token}", + "Content-Type": "application/json", + "Mcp-Protocol-Version": "2026-07-28", + "Mcp-Method": "tools/call", + "Mcp-Name": tool, + }, + ) + with urllib.request.urlopen(request, timeout=10) as response: + body = response.read().decode() + if ( + response.status != 200 + or '"error"' in body + or '"isError":true' in body.replace(" ", "") + ): + raise RuntimeError( + f"{url} {tool} failed: HTTP {response.status}: {body[:500]}" + ) + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("--urls", required=True) + parser.add_argument("--token-file", required=True) + args = parser.parse_args() + with open(args.token_file, encoding="utf-8") as stream: + token = stream.read().strip() + for url in args.urls.split(","): + for tool, arguments in TOOLS.items(): + call(url, token, tool, arguments) + print(f"PASS {url} {tool}") + + +if __name__ == "__main__": + main() diff --git a/benchmarks/fyre/report.py b/benchmarks/fyre/report.py new file mode 100644 index 0000000..24d364e --- /dev/null +++ b/benchmarks/fyre/report.py @@ -0,0 +1,182 @@ +"""Build machine-readable and Slack-ready FYRE scaling reports.""" + +from __future__ import annotations + +import argparse +import csv +import json +from pathlib import Path + +import matplotlib.pyplot as plt + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("--config", required=True) + parser.add_argument("--results", required=True) + args = parser.parse_args() + config = json.loads(Path(args.config).read_text(encoding="utf-8")) + results_root = Path(args.results) + results = {} + for scenario in config["scenarios"]: + path = results_root / scenario["id"] / "result.json" + if path.exists(): + results[scenario["id"]] = json.loads(path.read_text(encoding="utf-8")) + if "baseline" not in results: + raise RuntimeError("baseline result is required to calculate scaling") + + baseline = results["baseline"]["rps"] + rows = [] + for scenario in config["scenarios"]: + result = results.get(scenario["id"]) + if not result or result.get("status") != "confirmed": + continue + total_cpu = scenario["replicas"] * scenario["cpu"] + speedup = result["rps"] / baseline + row = { + "scenario": scenario["label"], + "scenario_id": scenario["id"], + "replicas": scenario["replicas"], + "cpu_per_vm": scenario["cpu"], + "memory_gb_per_vm": scenario["memory_gb"], + "total_cpu": total_cpu, + "total_memory_gb": scenario["replicas"] * scenario["memory_gb"], + "users": result["users"], + "lower_bound": result.get("lower_bound", False), + "rps": result["rps"], + "p50_ms": result["p50_ms"], + "p95_ms": result["p95_ms"], + "p99_ms": result["p99_ms"], + "speedup": speedup, + "efficiency": speedup / scenario["multiplier"], + "rps_per_vcpu": result["rps"] / total_cpu, + "rps_cv_percent": result["rps_cv_percent"], + "horizontal_advantage": None, + "replica_imbalance_percent": result["replica_imbalance_percent"], + } + if scenario["id"].startswith("horizontal-"): + vertical_id = scenario["id"].replace("horizontal", "vertical") + vertical = results.get(vertical_id) + if vertical and vertical.get("status") == "confirmed": + row["horizontal_advantage"] = result["rps"] / vertical["rps"] + rows.append(row) + + (results_root / "summary.json").write_text( + json.dumps({"rows": rows}, indent=2, sort_keys=True) + "\n", encoding="utf-8" + ) + with (results_root / "summary.csv").open( + "w", newline="", encoding="utf-8" + ) as stream: + writer = csv.DictWriter(stream, fieldnames=rows[0].keys()) + writer.writeheader() + writer.writerows(rows) + + figure = plt.figure(figsize=(16, 9), dpi=160, facecolor="#0b1020") + grid = figure.add_gridspec(2, 1, height_ratios=[2.1, 1.5], hspace=0.2) + axis = figure.add_subplot(grid[0]) + axis.set_facecolor("#0b1020") + colors = [ + "#a7b0c0" + if row["scenario_id"] == "baseline" + else "#49a7ff" + if "vertical" in row["scenario_id"] + else "#45d6a0" + for row in rows + ] + bars = axis.bar( + [row["scenario"] for row in rows], [row["rps"] for row in rows], color=colors + ) + axis.set_ylim(0, max(row["rps"] for row in rows) * 1.18) + axis.set_ylabel("Confirmed zero-error requests/second", color="white", fontsize=12) + axis.tick_params(axis="x", colors="white", rotation=12) + axis.tick_params(axis="y", colors="white") + for spine in axis.spines.values(): + spine.set_color("#56617a") + axis.grid(axis="y", color="#28324a", alpha=0.7) + for bar, row in zip(bars, rows): + comparison = f"{row['speedup']:.2f}x" + if row["horizontal_advantage"] is not None: + comparison += f"\n{row['horizontal_advantage']:.2f}x vs vertical" + axis.text( + bar.get_x() + bar.get_width() / 2, + bar.get_height(), + f"{row['rps']:,.0f} RPS\n{comparison}", + ha="center", + va="bottom", + color="white", + fontsize=10, + fontweight="bold", + ) + figure.suptitle( + "ContextForge Rust dataplane scaling on FYRE", + color="white", + fontsize=20, + fontweight="bold", + x=0.065, + y=0.985, + ha="left", + ) + figure.text( + 0.065, + 0.935, + "Same total dataplane CPU/RAM for matched vertical and horizontal comparisons", + color="#a7b0c0", + fontsize=10, + ha="left", + ) + figure.subplots_adjust(top=0.88) + + table_axis = figure.add_subplot(grid[1]) + table_axis.axis("off") + headers = [ + "Scenario", + "VMs × size", + "Users", + "RPS", + "p50 / p95 / p99 ms", + "Speedup", + "Efficiency", + "RPS/vCPU", + "CV / imbalance", + ] + cells = [] + for row in rows: + users = f"≥{row['users']:,}" if row["lower_bound"] else f"{row['users']:,}" + cells.append( + [ + row["scenario"], + f"{row['replicas']} × {row['cpu_per_vm']}c/{row['memory_gb_per_vm']}G", + users, + f"{row['rps']:,.0f}", + f"{row['p50_ms']:.1f} / {row['p95_ms']:.1f} / {row['p99_ms']:.1f}", + f"{row['speedup']:.2f}x", + f"{100 * row['efficiency']:.1f}%", + f"{row['rps_per_vcpu']:,.0f}", + f"{row['rps_cv_percent']:.1f}% / {row['replica_imbalance_percent']:.1f}%", + ] + ) + table = table_axis.table( + cellText=cells, + colLabels=headers, + cellLoc="center", + loc="center", + colWidths=[0.16, 0.12, 0.08, 0.09, 0.18, 0.09, 0.09, 0.10, 0.07], + ) + table.auto_set_font_size(False) + table.set_fontsize(9) + table.scale(1, 1.8) + for (row, _column), cell in table.get_celld().items(): + cell.set_edgecolor("#28324a") + cell.set_facecolor("#172039" if row else "#253250") + cell.get_text().set_color("white") + if row == 0: + cell.get_text().set_fontweight("bold") + figure.savefig( + results_root / "slack-scaling.png", + bbox_inches="tight", + facecolor=figure.get_facecolor(), + ) + + +if __name__ == "__main__": + main() diff --git a/benchmarks/fyre/scaling.yaml b/benchmarks/fyre/scaling.yaml new file mode 100644 index 0000000..51c1441 --- /dev/null +++ b/benchmarks/fyre/scaling.yaml @@ -0,0 +1,47 @@ +schema_version: 1 +infrastructure: + os: Ubuntu 24.04 + ssh_user: ubuntu + ssh_private_key: ~/.ssh/id_ed25519 + ssh_public_key: ~/.ssh/id_ed25519.pub + expiry_hours: 8 + helper_sizes: + - { cpu: 2, memory_gb: 8 } + - { cpu: 4, memory_gb: 16 } + - { cpu: 8, memory_gb: 32 } + - { cpu: 16, memory_gb: 32 } +images: + dataplane: ghcr.io/contextforge-org/contextforge-data-plane@sha256:0b3026b21659ce6c494142aee2c5866b44122232e73fcf21102da1f210692a96 + fast_time: ghcr.io/ibm/cfex-mcp-fast-time-server@sha256:110e1826f5d763e5afadba770b731dac93e0819c1bbadb68671b0124260603cf + helpers: ghcr.io/contextforge-org/cf-integration-helpers@sha256:f7e557e263737328225d827e73b519d0b40facb2698c55ee3d2de0ed82f00fa3 + locust: locustio/locust@sha256:fd39232c31971fe7509582717a95592f0fbf9a917b705f355bd7471e03bdd649 + redis: redis@sha256:a7859ed111db3c1f5404a973a4747505d559fb5ca32d37e447afc0ef845a2103 +workload: + protocol_version: 2026-07-28 + first_users: 125 + maximum_users: 32000 + ramp_seconds: 30 + warmup_seconds: 30 + measure_seconds: 120 + repetitions: 3 + maximum_campaign_seconds: 21600 + plateau_improvement_percent: 5.0 + boundary_percent: 12.5 + config_cache_seconds: 60 + helper_cpu_percent: 70.0 + helper_memory_percent: 80.0 + worker_core_percent: 85.0 + tools: + - convert_time + - echo + - get_stats + - get_system_time + - schema_success + - verify-protocol +scenarios: + - { id: baseline, label: Baseline, replicas: 1, cpu: 2, memory_gb: 8, multiplier: 1 } + - { id: vertical-2x, label: Vertical 2x, replicas: 1, cpu: 4, memory_gb: 16, multiplier: 2 } + - { id: horizontal-2x, label: Horizontal 2x, replicas: 2, cpu: 2, memory_gb: 8, multiplier: 2 } + - { id: vertical-3x, label: Vertical 3x, replicas: 1, cpu: 6, memory_gb: 24, multiplier: 3 } + - { id: horizontal-3x, label: Horizontal 3x, replicas: 3, cpu: 2, memory_gb: 8, multiplier: 3 } + - { id: vertical-4x, label: Vertical 4x extension, replicas: 1, cpu: 8, memory_gb: 32, multiplier: 4 } diff --git a/benchmarks/fyre/terraform/.terraform.lock.hcl b/benchmarks/fyre/terraform/.terraform.lock.hcl new file mode 100644 index 0000000..65b19c2 --- /dev/null +++ b/benchmarks/fyre/terraform/.terraform.lock.hcl @@ -0,0 +1,15 @@ +# This file is maintained automatically by "terraform init". +# Manual edits may be lost in future updates. + +provider "registry.terraform.io/hashicorp-forge/fyre" { + version = "0.0.3" + constraints = "0.0.3" + hashes = [ + "h1:rSh9ZK9JGe8duTe3ygRL5mbJeOmKDdosb/hA4P/5hB4=", + "zh:39052c8ec42ebd4862d00187226e8967b78f5c71d6d7472a6281852ba10e9097", + "zh:53c4a6fd9d58afe0cb52773137bbf23067606af92157e88a36425985ce9931c1", + "zh:5a17a4060432477ff00cfd9474ebff8fd4219908739466238fed99a12069334b", + "zh:890df766e9b839623b1f0437355032a3c006226a6c200cd911e15ee1a9014e9f", + "zh:c06c24f7fc4e9c4b1c2b06af530f89e9d0ec94b13efa9adad3956f6a4900e153", + ] +} diff --git a/benchmarks/fyre/terraform/main.tf b/benchmarks/fyre/terraform/main.tf new file mode 100644 index 0000000..0d0d47d --- /dev/null +++ b/benchmarks/fyre/terraform/main.tf @@ -0,0 +1,63 @@ +locals { + common = { + os = var.os + platform = "x" + public_network = "y" + quota_type = var.product_group_id == null ? "quick_burn" : "product_group" + product_group_id = var.product_group_id + site = var.site + ssh_keys = [var.ssh_public_key] + } +} + +resource "fyre_vm" "locust" { + hostname = "cf-${var.run_id}-locust" + description = "cf-integration FYRE benchmark ${var.run_id}; role=locust" + os = local.common.os + platform = local.common.platform + public_network = local.common.public_network + quota_type = local.common.quota_type + product_group_id = local.common.product_group_id + site = local.common.site + ssh_keys = local.common.ssh_keys + time_to_live = local.common.quota_type == "quick_burn" ? tostring(var.expiry_hours) : null + expiration = local.common.quota_type == "product_group" ? "${var.expiry_hours} hours" : null + cpu = var.locust_cpu + memory = var.locust_memory_gb + disable_delete = "n" +} + +resource "fyre_vm" "fast_time" { + hostname = "cf-${var.run_id}-fast-time" + description = "cf-integration FYRE benchmark ${var.run_id}; role=fast-time" + os = local.common.os + platform = local.common.platform + public_network = local.common.public_network + quota_type = local.common.quota_type + product_group_id = local.common.product_group_id + site = local.common.site + ssh_keys = local.common.ssh_keys + time_to_live = local.common.quota_type == "quick_burn" ? tostring(var.expiry_hours) : null + expiration = local.common.quota_type == "product_group" ? "${var.expiry_hours} hours" : null + cpu = var.fast_time_cpu + memory = var.fast_time_memory_gb + disable_delete = "n" +} + +resource "fyre_vm" "dataplane" { + count = var.dataplane_count + hostname = "cf-${var.run_id}-dataplane-${count.index + 1}" + description = "cf-integration FYRE benchmark ${var.run_id}; role=dataplane; replica=${count.index + 1}" + os = local.common.os + platform = local.common.platform + public_network = local.common.public_network + quota_type = local.common.quota_type + product_group_id = local.common.product_group_id + site = local.common.site + ssh_keys = local.common.ssh_keys + time_to_live = local.common.quota_type == "quick_burn" ? tostring(var.expiry_hours) : null + expiration = local.common.quota_type == "product_group" ? "${var.expiry_hours} hours" : null + cpu = var.dataplane_cpu + memory = var.dataplane_memory_gb + disable_delete = "n" +} diff --git a/benchmarks/fyre/terraform/outputs.tf b/benchmarks/fyre/terraform/outputs.tf new file mode 100644 index 0000000..77cf503 --- /dev/null +++ b/benchmarks/fyre/terraform/outputs.tf @@ -0,0 +1,37 @@ +locals { + locust_ips = { + for address in fyre_vm.locust.ips : address.type => address.ip + } + fast_time_ips = { + for address in fyre_vm.fast_time.ips : address.type => address.ip + } + dataplane_ips = [for vm in fyre_vm.dataplane : { + name = vm.hostname + id = vm.vm_id + ips = { for address in vm.ips : address.type => address.ip } + }] +} + +output "inventory" { + value = { + run_id = var.run_id + locust = { + name = fyre_vm.locust.hostname + id = fyre_vm.locust.vm_id + public_ip = try(local.locust_ips.public, "") + private_ip = try(local.locust_ips.private, "") + } + fast_time = { + name = fyre_vm.fast_time.hostname + id = fyre_vm.fast_time.vm_id + public_ip = try(local.fast_time_ips.public, "") + private_ip = try(local.fast_time_ips.private, "") + } + dataplanes = [for vm in local.dataplane_ips : { + name = vm.name + id = vm.id + public_ip = try(vm.ips.public, "") + private_ip = try(vm.ips.private, "") + }] + } +} diff --git a/benchmarks/fyre/terraform/variables.tf b/benchmarks/fyre/terraform/variables.tf new file mode 100644 index 0000000..1024747 --- /dev/null +++ b/benchmarks/fyre/terraform/variables.tf @@ -0,0 +1,31 @@ +variable "run_id" { + type = string + validation { + condition = can(regex("^[a-z0-9](?:[a-z0-9-]{0,46}[a-z0-9])?$", var.run_id)) + error_message = "run_id must be a lowercase DNS-safe identifier of at most 48 characters." + } +} + +variable "os" { type = string } +variable "ssh_public_key" { type = string } +variable "expiry_hours" { type = number } +variable "dataplane_count" { type = number } +variable "dataplane_cpu" { type = number } +variable "dataplane_memory_gb" { type = number } +variable "locust_cpu" { type = number } +variable "locust_memory_gb" { type = number } +variable "fast_time_cpu" { type = number } +variable "fast_time_memory_gb" { type = number } + +variable "product_group_id" { + type = string + default = null + nullable = true + sensitive = true +} + +variable "site" { + type = string + default = null + nullable = true +} diff --git a/benchmarks/fyre/terraform/versions.tf b/benchmarks/fyre/terraform/versions.tf new file mode 100644 index 0000000..f282187 --- /dev/null +++ b/benchmarks/fyre/terraform/versions.tf @@ -0,0 +1,11 @@ +terraform { + required_version = ">= 1.8, < 2.0" + required_providers { + fyre = { + source = "hashicorp-forge/fyre" + version = "= 0.0.3" + } + } +} + +provider "fyre" {} diff --git a/benchmarks/fyre/test_campaign.py b/benchmarks/fyre/test_campaign.py new file mode 100644 index 0000000..b5a92a1 --- /dev/null +++ b/benchmarks/fyre/test_campaign.py @@ -0,0 +1,289 @@ +"""Unit coverage for FYRE capacity-search and report-input behavior.""" + +from __future__ import annotations + +import csv +import json +import sys +import tempfile +import unittest +from pathlib import Path +from unittest import mock + +import campaign + +sys.path.insert(0, str(Path(__file__).parent / "deploy")) +import run_locust + + +def passed(users: int, rps: float) -> dict: + return { + "passed": True, + "users": users, + "rps": rps, + "failures": 0, + "p50_ms": 1.0, + "p95_ms": 2.0, + "p99_ms": 3.0, + "per_replica_rps": {"MCP tools/call [replica-1]": rps}, + "pressure": {}, + } + + +def config() -> dict: + return { + "workload": { + "first_users": 125, + "maximum_users": 32_000, + "maximum_campaign_seconds": 21_600, + "plateau_improvement_percent": 5.0, + "boundary_percent": 12.5, + "repetitions": 3, + "warmup_seconds": 30, + "measure_seconds": 120, + "helper_cpu_percent": 70.0, + "helper_memory_percent": 80.0, + "worker_core_percent": 85.0, + }, + "images": {"locust": "locust@sha256:test"}, + } + + +class CapacityTests(unittest.TestCase): + @mock.patch.object(campaign, "smoke") + @mock.patch.object(campaign, "one_phase") + @mock.patch.object(campaign, "measured_step") + def test_first_failure_never_advances_above_the_failed_load( + self, measured, phase, _smoke + ): + measured.side_effect = lambda _r, _c, _i, _u, _o, users, name: ( + passed(users, float(users)) + if name.startswith("confirm") or users <= 202 + else {"passed": False, "users": users, "reason": "first error"} + ) + phase.return_value = passed(202, 1000.0) + with tempfile.TemporaryDirectory() as directory: + result = campaign.capacity_search( + None, + config(), + { + "locust": {}, + "fast_time": {"private_ip": "10.0.0.2"}, + "dataplanes": [], + }, + [], + Path(directory), + ) + calls = [ + call.args[5] + for call in measured.call_args_list + if not call.args[6].startswith("confirm") + ] + first_failure = calls.index(250) + self.assertTrue(all(users <= 250 for users in calls[first_failure + 1 :])) + self.assertEqual(result["status"], "confirmed") + + @mock.patch.object(campaign, "smoke") + @mock.patch.object(campaign, "one_phase") + @mock.patch.object(campaign, "measured_step") + def test_two_sub_five_percent_steps_stop_at_plateau(self, measured, phase, _smoke): + rates = {125: 100.0, 250: 103.0, 500: 106.0} + measured.side_effect = lambda _r, _c, _i, _u, _o, users, _name: passed( + users, rates[users] + ) + phase.return_value = passed(500, 1000.0) + with tempfile.TemporaryDirectory() as directory: + result = campaign.capacity_search( + None, + config(), + { + "locust": {}, + "fast_time": {"private_ip": "10.0.0.2"}, + "dataplanes": [], + }, + [], + Path(directory), + ) + self.assertEqual(result["users"], 500) + self.assertNotIn(1000, [call.args[5] for call in measured.call_args_list]) + + def test_helper_saturation_uses_sustained_thresholds(self): + result = {"pressure": {"locust": {"mean_cpu_percent": 71.0}}} + self.assertEqual(campaign.helper_saturation(config(), result), "locust") + result["pressure"]["locust"]["mean_cpu_percent"] = 69.0 + self.assertIsNone(campaign.helper_saturation(config(), result)) + + @mock.patch.object(campaign, "smoke") + @mock.patch.object(campaign, "one_phase") + def test_warmup_and_measurement_are_separate_phases(self, phase, _smoke): + phase.side_effect = [passed(125, 90.0), passed(125, 100.0)] + result = campaign.measured_step( + None, config(), {"locust": {}}, [], Path("unused"), 125, "step" + ) + self.assertTrue(result["passed"]) + self.assertEqual( + [(call.args[7], call.args[6]) for call in phase.call_args_list], + [("step-warmup", 30), ("step", 120)], + ) + self.assertNotIn("measurement", phase.call_args_list[0].kwargs) + self.assertTrue(phase.call_args_list[1].kwargs["measurement"]) + + def test_pressure_excludes_ramp_and_warmup_samples(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "host.jsonl" + samples = [ + { + "kind": "sample", + "time": 100.0, + "cpu": { + "cpu": {"busy_percent": 100.0, "steal_percent": 30.0}, + "cpu0": {"busy_percent": 100.0, "steal_percent": 30.0}, + }, + "memory": {"used_percent": 99.0}, + "netstat": "TcpExt: ListenOverflows ListenDrops\nTcpExt: 0 0", + "docker_state": '{"Status":"running","OOMKilled":false}', + }, + { + "kind": "sample", + "time": 200.0, + "cpu": { + "cpu": {"busy_percent": 40.0, "steal_percent": 2.0}, + "cpu0": {"busy_percent": 45.0, "steal_percent": 2.0}, + }, + "memory": {"used_percent": 50.0}, + "netstat": "TcpExt: ListenOverflows ListenDrops\nTcpExt: 0 0", + "docker_state": '{"Status":"running","OOMKilled":false}', + }, + ] + path.write_text( + "".join(json.dumps(sample) + "\n" for sample in samples), + encoding="utf-8", + ) + result = campaign.pressure(path, after=150.0) + self.assertEqual(result["mean_cpu_percent"], 40.0) + self.assertEqual(result["max_memory_percent"], 50.0) + self.assertEqual(result["max_mean_core_percent"], 45.0) + self.assertEqual(result["mean_steal_percent"], 2.0) + self.assertFalse(result["worker_or_network_pressure"]) + + def test_pressure_detects_network_drops(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "host.jsonl" + samples = [] + for timestamp, drops in ((100.0, 0), (101.0, 1)): + samples.append( + { + "kind": "sample", + "time": timestamp, + "cpu": {}, + "memory": {"used_percent": 10.0}, + "netstat": ( + "TcpExt: ListenOverflows ListenDrops TCPBacklogDrop\n" + f"TcpExt: 0 {drops} 0" + ), + "docker_state": "", + } + ) + path.write_text( + "".join(json.dumps(sample) + "\n" for sample in samples), + encoding="utf-8", + ) + result = campaign.pressure(path) + self.assertTrue(result["worker_or_network_pressure"]) + + def test_docker_pressure_ignores_clean_exit_and_detects_oom(self): + self.assertFalse( + campaign.docker_pressure( + '{"Status":"exited","ExitCode":0,"OOMKilled":false}' + ) + ) + self.assertTrue( + campaign.docker_pressure( + '{"Status":"exited","ExitCode":137,"OOMKilled":true}' + ) + ) + + @mock.patch.object(run_locust.time, "sleep") + @mock.patch.object(run_locust, "docker") + @mock.patch.object(run_locust, "container_state") + def test_worker_exit_stops_the_master_immediately(self, state, docker, _sleep): + state.side_effect = [("running", 0), ("exited", 2)] + self.assertEqual(run_locust.wait_for_cluster("master", ["worker"]), 1) + docker.assert_called_once_with( + "stop", "--time", "1", "master", check=False, capture=True + ) + + def test_stats_preserve_replica_rates_and_exclude_discovery(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "stats.csv" + fields = [ + "Type", + "Name", + "Request Count", + "Failure Count", + "Requests/s", + "50%", + "95%", + "99%", + ] + with path.open("w", newline="", encoding="utf-8") as stream: + writer = csv.DictWriter(stream, fieldnames=fields) + writer.writeheader() + writer.writerow( + { + "Name": "MCP server/discover", + "Request Count": 100, + "Failure Count": 0, + "Requests/s": 50, + "50%": 50, + "95%": 80, + "99%": 90, + } + ) + writer.writerow( + { + "Name": "MCP tools/call [replica-1]", + "Request Count": 1000, + "Failure Count": 0, + "Requests/s": 500, + "50%": 2, + "95%": 4, + "99%": 5, + } + ) + writer.writerow( + { + "Name": "MCP tools/call [replica-2]", + "Request Count": 900, + "Failure Count": 0, + "Requests/s": 450, + "50%": 3, + "95%": 5, + "99%": 7, + } + ) + writer.writerow( + { + "Name": "Aggregated", + "Request Count": 1900, + "Failure Count": 0, + "Requests/s": 950, + "50%": 2.5, + "95%": 4.5, + "99%": 6, + } + ) + result = campaign.read_stats(path) + aggregate = campaign.read_stats(path, use_aggregate=True) + self.assertEqual(result["requests"], 1900) + self.assertEqual(result["rps"], 950.0) + self.assertEqual(len(result["per_replica_rps"]), 2) + self.assertLess(result["p95_ms"], 5.0) + + self.assertEqual(aggregate["p50_ms"], 2.5) + self.assertEqual(aggregate["p95_ms"], 4.5) + self.assertEqual(aggregate["p99_ms"], 6.0) + + +if __name__ == "__main__": + unittest.main() diff --git a/docker/helpers.Dockerfile b/docker/helpers.Dockerfile index 154f184..0e35752 100644 --- a/docker/helpers.Dockerfile +++ b/docker/helpers.Dockerfile @@ -4,6 +4,7 @@ COPY Cargo.toml Cargo.lock ./ COPY src ./src COPY docker ./docker COPY scripts ./scripts +COPY benchmarks ./benchmarks COPY tests/conformance/baselines ./tests/conformance/baselines RUN --mount=type=cache,target=/usr/local/cargo/registry \ --mount=type=cache,target=/usr/local/cargo/git \ diff --git a/scripts/locustfile_mcp.py b/scripts/locustfile_mcp.py index 4ccfd82..d9cd824 100644 --- a/scripts/locustfile_mcp.py +++ b/scripts/locustfile_mcp.py @@ -12,27 +12,44 @@ MCPGATEWAY_BEARER_TOKEN bearer token (required) MCP_TOOL_NAMES optional comma-separated tools to call MCP_SKIP_TOOL_LIST true when direct tool aliases are supplied + MCP_BASE_URLS optional comma-separated replica origins + MCP_DIRECT_DATAPLANE use the native dataplane route without nginx + MCP_FYRE_WORKLOAD enable the six-tool FYRE workload arguments + MCP_EXPLICIT_ZERO_DELAY send zero delay to Fast Time echo + MCP_MEASUREMENT_MARKER FYRE path written when spawning completes + MCP_MEASUREMENT_SECONDS FYRE measured duration after the marker LOCUST_REQUEST_TIMEOUT_SECONDS positive finite per-request timeout (default 60) """ + from __future__ import annotations +import itertools import json import logging import math import os import random +import time import uuid +from pathlib import Path from urllib.parse import quote import gevent -from locust import HttpUser, constant, events, task +from locust import constant, events, task + +try: + from locust import FastHttpUser +except ImportError: # Minimal test doubles expose only HttpUser. + from locust import HttpUser as FastHttpUser from locust.runners import MasterRunner, WorkerRunner PROTOCOL_VERSION = os.environ.get("MCP_PROTOCOL_VERSION", "2026-07-28") STATELESS = PROTOCOL_VERSION == "2026-07-28" LEGACY_PROTOCOL_VERSIONS = {"2024-11-05", "2025-03-26", "2025-06-18", "2025-11-25"} if PROTOCOL_VERSION not in {"2025-11-25", "2026-07-28"}: - raise RuntimeError("MCP_PROTOCOL_VERSION must be a harness-selected client revision") + raise RuntimeError( + "MCP_PROTOCOL_VERSION must be a harness-selected client revision" + ) ACCEPT = "application/json, text/event-stream" _REQUEST_TIMEOUT_ERROR = ( "LOCUST_REQUEST_TIMEOUT_SECONDS must be a finite number greater than zero" @@ -58,6 +75,17 @@ def _request_timeout_seconds() -> float: "fast_time_echo": {"message": "cf-integration"}, "fast-time-echo": {"message": "cf-integration"}, } +_FYRE_TOOL_ARGUMENTS = { + "convert_time": { + "time": "12:00", + "source_timezone": "UTC", + "target_timezone": "Europe/Dublin", + }, + "get_stats": {}, + "get_system_time": {"timezone": "UTC"}, + "schema_success": {}, + "verify-protocol": {}, +} def jsonrpc(method: str, params: dict | None = None) -> dict: @@ -124,7 +152,20 @@ def parse_mcp_body(text: str, content_type: str): def tool_call_args(tool_name: str) -> dict | None: """Use the same Fast Time echo payload for raw and control-plane aliases.""" arguments = _TOOL_ARGUMENTS.get(tool_name) - return dict(arguments) if arguments is not None else None + if ( + arguments is None + and os.environ.get("MCP_FYRE_WORKLOAD", "false").lower() == "true" + ): + arguments = _FYRE_TOOL_ARGUMENTS.get(tool_name) + if arguments is None: + return None + result = dict(arguments) + if ( + tool_name == "echo" + and os.environ.get("MCP_EXPLICIT_ZERO_DELAY", "false").lower() == "true" + ): + result["delay"] = 0 + return result def validate_result(method: str, result) -> dict: @@ -134,7 +175,9 @@ def validate_result(method: str, result) -> dict: if method == "initialize": version = result.get("protocolVersion") if not isinstance(version, str) or version not in LEGACY_PROTOCOL_VERSIONS: - raise ValueError("initialize must negotiate a supported legacy protocol revision") + raise ValueError( + "initialize must negotiate a supported legacy protocol revision" + ) if not isinstance(result.get("capabilities"), dict): raise ValueError("initialize result must include capabilities") server_info = result.get("serverInfo") @@ -142,11 +185,15 @@ def validate_result(method: str, result) -> dict: isinstance(server_info.get(field), str) and server_info[field] for field in ("name", "version") ): - raise ValueError("initialize result must include serverInfo name and version") + raise ValueError( + "initialize result must include serverInfo name and version" + ) elif method == "server/discover": versions = result.get("supportedVersions") if not isinstance(versions, list) or PROTOCOL_VERSION not in versions: - raise ValueError("server/discover must advertise the requested protocol version") + raise ValueError( + "server/discover must advertise the requested protocol version" + ) if not isinstance(result.get("capabilities"), dict): raise ValueError("server/discover result must include capabilities") if not isinstance(result.get("resultType"), str): @@ -184,11 +231,23 @@ def validate_result(method: str, result) -> dict: raise ValueError("tools/call result contains invalid content") return result + MCP_SERVER_ID = os.environ.get("MCP_SERVER_ID", "") MCP_STACK_MODE = os.environ.get("MCP_STACK_MODE", "dataplane") BEARER_TOKEN = os.environ.get("MCPGATEWAY_BEARER_TOKEN", "") -TOOL_NAMES = [name.strip() for name in os.environ.get("MCP_TOOL_NAMES", "").split(",") if name.strip()] +TOOL_NAMES = [ + name.strip() + for name in os.environ.get("MCP_TOOL_NAMES", "").split(",") + if name.strip() +] SKIP_TOOL_LIST = os.environ.get("MCP_SKIP_TOOL_LIST", "false").lower() == "true" +BASE_URLS = [ + url.strip().rstrip("/") + for url in os.environ.get("MCP_BASE_URLS", "").split(",") + if url.strip() +] +DIRECT_DATAPLANE = os.environ.get("MCP_DIRECT_DATAPLANE", "false").lower() == "true" +_TARGET_SEQUENCE = itertools.count() def safe_diagnostic(value) -> str: @@ -201,6 +260,8 @@ def mcp_path() -> str: """Return the mode-aware public MCP route.""" if MCP_STACK_MODE == "controlplane": return "/mcp" + if DIRECT_DATAPLANE: + return f"/contextforge-rs/servers/{quote(MCP_SERVER_ID, safe='')}/mcp" return f"/servers/{quote(MCP_SERVER_ID, safe='')}/mcp" @@ -226,6 +287,15 @@ def stop_from_worker(msg=None, **_message): if isinstance(environment.runner, MasterRunner): environment.runner.register_message(_FAIL_FAST_MESSAGE, stop_from_worker) + marker = os.environ.get("MCP_MEASUREMENT_MARKER") + if marker: + + def mark_measurement_start(_user_count: int) -> None: + Path(marker).write_text(f"{time.time()}\n", encoding="utf-8") + seconds = float(os.environ["MCP_MEASUREMENT_SECONDS"]) + gevent.spawn_later(seconds, environment.runner.quit) + + environment.events.spawning_complete.add_listener(mark_measurement_start) def stop_on_error(exception=None, **_kwargs): nonlocal stopping @@ -256,12 +326,18 @@ def fail_empty_run(environment, **_kwargs) -> None: environment.process_exit_code = 1 -class MCPGatewayUser(HttpUser): +class MCPGatewayUser(FastHttpUser): """Drives discovery or initialization, then tool requests on the public route.""" wait_time = constant(0) + host = BASE_URLS[0] if BASE_URLS else None def __init__(self, *args, **kwargs): + self._replica_index = ( + next(_TARGET_SEQUENCE) % len(BASE_URLS) if BASE_URLS else None + ) + if self._replica_index is not None: + self.host = BASE_URLS[self._replica_index] super().__init__(*args, **kwargs) self._session_id: str | None = None self._protocol_version = PROTOCOL_VERSION @@ -300,7 +376,9 @@ def on_start(self): raise RuntimeError("initialize response did not include Mcp-Session-Id") if not STATELESS: self._protocol_version = result["protocolVersion"] - if not self._mcp_notification("notifications/initialized", None, name="MCP initialized"): + if not self._mcp_notification( + "notifications/initialized", None, name="MCP initialized" + ): return if not self._tool_names and not SKIP_TOOL_LIST: listed = self._mcp_request("tools/list", {}, name="MCP tools/list") @@ -312,9 +390,13 @@ def on_start(self): and isinstance(tool.get("name"), str) and tool["name"].strip() ] - self._tool_names = [name for name in self._tool_names if tool_call_args(name) is not None] + self._tool_names = [ + name for name in self._tool_names if tool_call_args(name) is not None + ] if not self._tool_names: - raise RuntimeError("Fast Time echo tool is required; refusing an empty load workload") + raise RuntimeError( + "Fast Time echo tool is required; refusing an empty load workload" + ) self._ready = True def on_stop(self): @@ -331,7 +413,9 @@ def on_stop(self): if not self._validate_backend(response): return if response.status_code not in (200, 202, 204, 404, 405): - response.failure(f"HTTP {response.status_code}; expected session termination response") + response.failure( + f"HTTP {response.status_code}; expected session termination response" + ) return response.success() @@ -371,9 +455,13 @@ def _headers( @staticmethod def _validate_backend(response) -> bool: - if MCP_STACK_MODE != "dataplane": + if MCP_STACK_MODE != "dataplane" or DIRECT_DATAPLANE: return True - marker = response.headers.get("X-CF-Integration-Backend") if response.headers else None + marker = ( + response.headers.get("X-CF-Integration-Backend") + if response.headers + else None + ) if marker != "dataplane": response.failure("Missing or invalid dataplane backend marker") return False @@ -405,16 +493,25 @@ def _mcp_request( ) as response: if not self._validate_backend(response): return None - session_id = response.headers.get("Mcp-Session-Id") if response.headers else None + session_id = ( + response.headers.get("Mcp-Session-Id") if response.headers else None + ) if session_id and not STATELESS: self._session_id = session_id if response.status_code != 200: detail = getattr(response, "error", None) - response.failure(safe_diagnostic(f"HTTP {response.status_code}" + (f": {detail}" if detail else ""))) + response.failure( + safe_diagnostic( + f"HTTP {response.status_code}" + + (f": {detail}" if detail else "") + ) + ) return None try: - message = parse_mcp_body(response.text, response.headers.get("Content-Type", "")) + message = parse_mcp_body( + response.text, response.headers.get("Content-Type", "") + ) except ValueError as exc: response.failure(safe_diagnostic(f"Invalid body: {exc}")) return None @@ -460,7 +557,12 @@ def _mcp_notification(self, method: str, params: dict | None, name: str) -> bool return False if response.status_code != 202: detail = getattr(response, "error", None) - response.failure(safe_diagnostic(f"HTTP {response.status_code}; expected 202" + (f": {detail}" if detail else ""))) + response.failure( + safe_diagnostic( + f"HTTP {response.status_code}; expected 202" + + (f": {detail}" if detail else "") + ) + ) return False if response.content: response.failure("HTTP 202 notification response body must be empty") @@ -474,4 +576,7 @@ def tools_call(self): return tool = random.choice(self._tool_names) args = tool_call_args(tool) - self._mcp_request("tools/call", {"name": tool, "arguments": args}, name="MCP tools/call") + name = "MCP tools/call" + if self._replica_index is not None: + name += f" [replica-{self._replica_index + 1}]" + self._mcp_request("tools/call", {"name": tool, "arguments": args}, name=name) diff --git a/src/app.rs b/src/app.rs index 137d37c..110aa39 100644 --- a/src/app.rs +++ b/src/app.rs @@ -15,7 +15,7 @@ use crate::performance::LoadRequest; use anyhow::{Result, bail}; use crate::cli::{ - CiCommand, Cli, CliLane, CliRoutedLane, Command, ConformanceCommand, DebugCommand, + CiCommand, Cli, CliLane, CliRoutedLane, Command, ConformanceCommand, DebugCommand, FyreCommand, LaneSelection, LiveGroup, LoadCommand, ProtocolVersion, StackCommand, TokenKind, }; const LANE_ENV: &str = "CF_MCP_LANE"; @@ -31,6 +31,7 @@ pub(crate) enum Action { protocol_version: ProtocolVersion, }, Load(ResolvedLoadArgs), + Fyre(FyreAction), Live { lane: SemanticLane, group: LiveGroup, @@ -53,6 +54,9 @@ impl Action { Self::Stack(StackAction::Config { .. }) => "stack config", Self::Probe { .. } => "probe", Self::Load(_) => "load test", + Self::Fyre(FyreAction::Run { .. }) => "FYRE scaling benchmark", + Self::Fyre(FyreAction::Status { .. }) => "FYRE benchmark status", + Self::Fyre(FyreAction::Destroy { .. }) => "FYRE benchmark destroy", Self::Live { .. } => "live tests", Self::Conformance(ConformanceAction::Run { .. }) => "conformance tests", Self::Conformance(ConformanceAction::Report { .. }) => "conformance report", @@ -101,6 +105,7 @@ impl Action { } summary } + Self::Fyre(action) => action.startup_summary(), Self::Live { lane, protocol_version, @@ -155,6 +160,7 @@ impl Action { self, Self::Stack(StackAction::Up { .. }) | Self::Load(_) + | Self::Fyre(FyreAction::Run { .. }) | Self::Conformance(ConformanceAction::Run { .. }) ) } @@ -180,6 +186,7 @@ impl Action { | Self::Stack(StackAction::Logs { standalone, .. }) | Self::Stack(StackAction::Config { standalone, .. }) => *standalone, Self::Load(args) => args.standalone, + Self::Fyre(_) => true, Self::Live { .. } => false, }; if standalone { @@ -190,6 +197,32 @@ impl Action { } } +/// A resolved operation on one FYRE benchmark run. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) enum FyreAction { + Run { + file: Option, + run_id: Option, + }, + Status { + run_id: String, + }, + Destroy { + run_id: String, + }, +} + +impl FyreAction { + fn startup_summary(&self) -> String { + let (operation, run_id) = match self { + Self::Run { run_id, .. } => ("run", run_id.as_deref().unwrap_or("generated")), + Self::Status { run_id, .. } => ("status", run_id.as_str()), + Self::Destroy { run_id, .. } => ("destroy", run_id.as_str()), + }; + format!("Infrastructure: FYRE\nOperation: {operation}\nRun ID: {run_id}") + } +} + impl StackAction { fn startup_summary(&self) -> String { let lane = match self { @@ -400,29 +433,49 @@ pub(crate) fn resolve_action(cli: Cli, environment: &Environment) -> Result { - let LoadCommand::Run(args) = args.command; - let topology = resolve_lane(args.lane, environment)?; - validate_standalone_lane(standalone, topology)?; - if args.builtin_memory_limit.is_some() && topology != StackMode::Controlplane { - bail!("--builtin-memory-limit requires --lane builtin"); + Command::Load(args) => match args.command { + LoadCommand::Run(args) => { + let topology = resolve_lane(args.lane, environment)?; + validate_standalone_lane(standalone, topology)?; + if args.builtin_memory_limit.is_some() && topology != StackMode::Controlplane { + bail!("--builtin-memory-limit requires --lane builtin"); + } + Ok(Action::Load(ResolvedLoadArgs { + topology, + client_era: args.client_era, + standalone, + observability: args.observability, + builtin_memory_limit: args.builtin_memory_limit, + isolate_cpus: args.isolate_cpus, + request: LoadRequest { + smoke: args.smoke, + users: args.users, + spawn_rate: args.spawn_rate, + run_time: args.run_time, + workers: args.workers, + }, + })) } - Ok(Action::Load(ResolvedLoadArgs { - topology, - client_era: args.client_era, - standalone, - observability: args.observability, - builtin_memory_limit: args.builtin_memory_limit, - isolate_cpus: args.isolate_cpus, - request: LoadRequest { - smoke: args.smoke, - users: args.users, - spawn_rate: args.spawn_rate, - run_time: args.run_time, - workers: args.workers, - }, - })) - } + LoadCommand::Fyre(args) => { + if standalone { + bail!( + "--standalone is not used by load fyre; FYRE targets are always isolated" + ); + } + Ok(Action::Fyre(match args.command { + FyreCommand::Run(args) => FyreAction::Run { + file: args.file, + run_id: args.run_id, + }, + FyreCommand::Status(args) => FyreAction::Status { + run_id: validated_run_id(args.run_id)?, + }, + FyreCommand::Destroy(args) => FyreAction::Destroy { + run_id: validated_run_id(args.run_id)?, + }, + })) + } + }, Command::Live(args) => { let lane = resolve_live_lane(args.target.lane, environment)?; if standalone { @@ -539,6 +592,30 @@ pub(crate) fn resolve_action(cli: Cli, environment: &Environment) -> Result Result { + if is_valid_run_id(&run_id) { + Ok(run_id) + } else { + bail!("--run-id must contain only lowercase letters, digits, and hyphens") + } +} + +fn is_valid_run_id(run_id: &str) -> bool { + !run_id.is_empty() + && run_id.len() <= 48 + && run_id + .bytes() + .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || byte == b'-') + && run_id + .as_bytes() + .first() + .is_some_and(u8::is_ascii_alphanumeric) + && run_id + .as_bytes() + .last() + .is_some_and(u8::is_ascii_alphanumeric) +} + fn environment_utf8(environment: &Environment, key: &str) -> Option { environment .get(std::ffi::OsStr::new(key)) diff --git a/src/app_tests.rs b/src/app_tests.rs index b7af5c7..37a6d13 100644 --- a/src/app_tests.rs +++ b/src/app_tests.rs @@ -2,7 +2,8 @@ use std::ffi::OsString; use std::path::PathBuf; use cf_integration::app::{ - Action, CiAction, ConformanceAction, DebugAction, ResolvedLoadArgs, StackAction, resolve_action, + Action, CiAction, ConformanceAction, DebugAction, FyreAction, ResolvedLoadArgs, StackAction, + resolve_action, }; use cf_integration::cli::{Cli, LaneSelection, LiveGroup, ProtocolVersion, TokenKind}; use cf_integration::conformance::results::{ConformanceServerEra, SemanticLane}; @@ -30,6 +31,32 @@ fn every_subcommand_has_a_stable_progress_description() { (&["cf-integration", "stack", "config"], "stack config"), (&["cf-integration", "probe"], "probe"), (&["cf-integration", "load", "run"], "load test"), + ( + &["cf-integration", "load", "fyre", "run"], + "FYRE scaling benchmark", + ), + ( + &[ + "cf-integration", + "load", + "fyre", + "status", + "--run-id", + "scale-run", + ], + "FYRE benchmark status", + ), + ( + &[ + "cf-integration", + "load", + "fyre", + "destroy", + "--run-id", + "scale-run", + ], + "FYRE benchmark destroy", + ), (&["cf-integration", "live"], "live tests"), ( &["cf-integration", "conformance", "run"], @@ -201,11 +228,54 @@ fn conformance_startup_labels_both_legacy_era_selections() { fn multi_phase_commands_own_detailed_progress_while_simple_commands_use_global_progress() { assert!(!action(&["cf-integration", "stack", "up"], &[]).uses_global_activity()); assert!(!action(&["cf-integration", "load", "run"], &[]).uses_global_activity()); + assert!(!action(&["cf-integration", "load", "fyre", "run"], &[]).uses_global_activity()); assert!(!action(&["cf-integration", "conformance", "run"], &[]).uses_global_activity()); assert!(action(&["cf-integration", "stack", "down"], &[]).uses_global_activity()); assert!(action(&["cf-integration", "probe"], &[]).uses_global_activity()); } +#[test] +fn fyre_actions_are_isolated_runtime_operations() { + assert_eq!( + action( + &[ + "cf-integration", + "load", + "fyre", + "run", + "--file", + "matrix.yaml", + "--run-id", + "scale-run", + ], + &[], + ), + Action::Fyre(FyreAction::Run { + file: Some(PathBuf::from("matrix.yaml")), + run_id: Some("scale-run".to_owned()), + }) + ); + let status = action( + &[ + "cf-integration", + "load", + "fyre", + "status", + "--run-id", + "scale-run", + ], + &[], + ); + assert_eq!( + status.config_requirements(), + ConfigRequirements::StandaloneRuntime + ); + assert_eq!( + status.startup_summary(), + "Infrastructure: FYRE\nOperation: status\nRun ID: scale-run" + ); +} + #[test] fn lane_precedence_is_cli_then_environment_then_external() { assert_eq!( diff --git a/src/cli.rs b/src/cli.rs index 8c0a727..6f485f1 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -323,6 +323,51 @@ pub(crate) enum LoadCommand { /// Run Locust through the selected public MCP route. #[command(visible_alias = "r")] Run(LoadRunArgs), + /// Run repeatable scaling benchmarks on FYRE virtual machines. + #[command(visible_alias = "f")] + Fyre(FyreArgs), +} + +/// FYRE benchmark command selection. +#[derive(Debug, Clone, PartialEq, Eq, Args)] +pub(crate) struct FyreArgs { + /// FYRE benchmark operation to run. + #[command(subcommand)] + pub(crate) command: FyreCommand, +} + +/// Operations on one FYRE benchmark run. +#[derive(Debug, Clone, PartialEq, Eq, Subcommand)] +pub(crate) enum FyreCommand { + /// Provision, benchmark, download reports, and destroy run-owned VMs. + #[command(visible_alias = "r")] + Run(FyreRunArgs), + /// Show durable state for a benchmark run. + #[command(visible_alias = "s")] + Status(FyreExistingRunArgs), + /// Destroy only the VMs owned by a benchmark run. + #[command(visible_alias = "d")] + Destroy(FyreExistingRunArgs), +} + +/// Common FYRE benchmark options. +#[derive(Debug, Clone, PartialEq, Eq, Args)] +pub(crate) struct FyreRunArgs { + /// Scenario configuration file; defaults to the packaged scaling matrix. + #[arg(short = 'f', long, value_name = "FILE")] + pub(crate) file: Option, + + /// Run identifier; generated when omitted. + #[arg(short = 'i', long, value_name = "RUN_ID")] + pub(crate) run_id: Option, +} + +/// Options for an existing FYRE benchmark run. +#[derive(Debug, Clone, PartialEq, Eq, Args)] +pub(crate) struct FyreExistingRunArgs { + /// Existing run identifier. + #[arg(short = 'i', long, value_name = "RUN_ID", required = true)] + pub(crate) run_id: String, } /// Load-test options. diff --git a/src/cli_public_tests.rs b/src/cli_public_tests.rs index 8a7154c..1f1f60b 100644 --- a/src/cli_public_tests.rs +++ b/src/cli_public_tests.rs @@ -70,7 +70,7 @@ fn command_tree_contains_only_distinct_public_workflows() { subcommands(&["stack"]), ["up", "down", "status", "logs", "config"] ); - assert_eq!(subcommands(&["load"]), ["run"]); + assert_eq!(subcommands(&["load"]), ["run", "fyre"]); assert_eq!(subcommands(&["conformance"]), ["run", "report"]); assert_eq!(subcommands(&["debug"]), ["inspect", "token"]); } @@ -88,6 +88,10 @@ fn every_public_command_renders_help() { &["probe"], &["load"], &["load", "run"], + &["load", "fyre"], + &["load", "fyre", "run"], + &["load", "fyre", "status"], + &["load", "fyre", "destroy"], &["live"], &["conformance"], &["conformance", "run"], @@ -282,7 +286,9 @@ fn load_accepts_standalone_external_dataplane_mode() { panic!("expected load") }; - let LoadCommand::Run(args) = args.command; + let LoadCommand::Run(args) = args.command else { + panic!("expected load run") + }; assert_eq!(args.lane, Some(CliRoutedLane::External)); } @@ -293,7 +299,9 @@ fn load_accepts_explicit_observability() { panic!("expected load") }; - let LoadCommand::Run(args) = args.command; + let LoadCommand::Run(args) = args.command else { + panic!("expected load run") + }; assert!(args.observability); } @@ -695,7 +703,9 @@ fn load_uses_client_eras_and_rejects_version_or_server_selectors() { else { panic!("expected load") }; - let LoadCommand::Run(args) = args.command; + let LoadCommand::Run(args) = args.command else { + panic!("expected load run") + }; assert_eq!(args.client_era, expected); } for arguments in [ @@ -838,6 +848,26 @@ fn short_commands_and_options_resolve_identically_to_long_forms() { "16G", ], ), + ( + &["l", "f", "r", "-f", "scenario.yaml", "-i", "scale-run"], + &[ + "load", + "fyre", + "run", + "--file", + "scenario.yaml", + "--run-id", + "scale-run", + ], + ), + ( + &["l", "f", "s", "-i", "scale-run"], + &["load", "fyre", "status", "--run-id", "scale-run"], + ), + ( + &["l", "f", "d", "-i", "scale-run"], + &["load", "fyre", "destroy", "--run-id", "scale-run"], + ), ( &["v", "-l", "builtin", "-p", "legacy", "-g", "protocol"], &[ diff --git a/src/infrastructure/assets.rs b/src/infrastructure/assets.rs index f9ce327..4bf631e 100644 --- a/src/infrastructure/assets.rs +++ b/src/infrastructure/assets.rs @@ -31,6 +31,20 @@ static ASSETS: LazyLock> = LazyLock::new(|| { let mut assets = vec![ asset!("Cargo.toml"), asset!("Cargo.lock"), + asset!("benchmarks/fyre/scaling.yaml"), + asset!("benchmarks/fyre/campaign.py"), + asset!("benchmarks/fyre/report.py"), + asset!("benchmarks/fyre/README.md"), + asset!("benchmarks/fyre/deploy/dataplane.compose.yaml"), + asset!("benchmarks/fyre/deploy/fast-time.compose.yaml"), + asset!("benchmarks/fyre/deploy/monitor.py"), + asset!("benchmarks/fyre/deploy/run_locust.py"), + asset!("benchmarks/fyre/deploy/smoke.py"), + asset!("benchmarks/fyre/terraform/main.tf"), + asset!("benchmarks/fyre/terraform/.terraform.lock.hcl"), + asset!("benchmarks/fyre/terraform/outputs.tf"), + asset!("benchmarks/fyre/terraform/variables.tf"), + asset!("benchmarks/fyre/terraform/versions.tf"), asset!("docker/clickstack/collector.yaml"), asset!("docker/docker-compose.cf-conformance-fixture.yaml"), asset!("docker/docker-compose.cf-conformance-controlplane.yaml"), diff --git a/src/runtime/fyre.rs b/src/runtime/fyre.rs new file mode 100644 index 0000000..9a12eab --- /dev/null +++ b/src/runtime/fyre.rs @@ -0,0 +1,826 @@ +//! Repeatable FYRE infrastructure and scaling-campaign orchestration. + +use std::collections::BTreeSet; +use std::ffi::{OsStr, OsString}; +use std::fs; +use std::path::{Path, PathBuf}; +use std::time::Duration; + +use anyhow::{Context, Result, bail, ensure}; +use serde::{Deserialize, Serialize}; +use serde_json::{Value, json}; +use uuid::Uuid; + +use super::{AppFailure, AppResult, CommandSpec, ProcessRunner, RuntimeContext}; +use crate::app::FyreAction; + +const OWNERSHIP_FILE: &str = "run.json"; +const TERRAFORM_DIRECTORY: &str = "terraform"; +const TERRAFORM_VARIABLES: &str = "scenario.tfvars.json"; +const HELPER_SATURATION_EXIT: i32 = 42; + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct FyreConfig { + schema_version: u32, + infrastructure: InfrastructureConfig, + images: ImageConfig, + workload: WorkloadConfig, + scenarios: Vec, + #[serde(skip_serializing_if = "Option::is_none")] + active_helper: Option, + #[serde(skip_serializing_if = "Option::is_none")] + resolved_ssh_private_key: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct InfrastructureConfig { + os: String, + ssh_user: String, + ssh_private_key: PathBuf, + ssh_public_key: PathBuf, + expiry_hours: u32, + helper_sizes: Vec, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize)] +struct MachineSize { + cpu: u32, + memory_gb: u32, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct ActiveHelper { + locust_cpu: u32, + locust_memory_gb: u32, + fast_time_cpu: u32, + fast_time_memory_gb: u32, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct ImageConfig { + dataplane: String, + fast_time: String, + helpers: String, + locust: String, + redis: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct WorkloadConfig { + protocol_version: String, + first_users: u32, + maximum_users: u32, + ramp_seconds: u32, + warmup_seconds: u32, + measure_seconds: u32, + repetitions: u32, + maximum_campaign_seconds: u32, + plateau_improvement_percent: f64, + boundary_percent: f64, + config_cache_seconds: u32, + helper_cpu_percent: f64, + helper_memory_percent: f64, + worker_core_percent: f64, + tools: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct Scenario { + id: String, + label: String, + replicas: u32, + cpu: u32, + memory_gb: u32, + multiplier: u32, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct RunState { + schema_version: u32, + run_id: String, + phase: String, + config_file: PathBuf, + current_scenario: Option, + locust_helper_size: usize, + fast_time_helper_size: usize, + completed_scenarios: Vec, + cleanup_required: bool, +} + +impl RuntimeContext { + pub(super) async fn execute_fyre(&self, action: FyreAction) -> AppResult<()> { + match action { + FyreAction::Run { file, run_id } => self.run_fyre(file, run_id).await, + FyreAction::Status { run_id } => self.fyre_status(&run_id), + FyreAction::Destroy { run_id } => self.destroy_fyre(&run_id).await, + } + } + + async fn run_fyre(&self, file: Option, run_id: Option) -> AppResult<()> { + let source = file.unwrap_or_else(|| { + self.config + .asset_root() + .join("benchmarks/fyre/scaling.yaml") + }); + let mut config = read_config(&source).map_err(AppFailure::from)?; + validate_config(&config).map_err(AppFailure::from)?; + self.require_fyre_credentials()?; + let private_key = + expand_home(&config.infrastructure.ssh_private_key).map_err(AppFailure::from)?; + let public_key = + expand_home(&config.infrastructure.ssh_public_key).map_err(AppFailure::from)?; + ensure_file(&private_key, "SSH private key").map_err(AppFailure::from)?; + ensure_file(&public_key, "SSH public key").map_err(AppFailure::from)?; + config.resolved_ssh_private_key = Some(private_key); + + let run_id = run_id + .unwrap_or_else(|| format!("scale-{}", &Uuid::new_v4().simple().to_string()[..12])); + validate_run_id(&run_id).map_err(AppFailure::from)?; + let root = self.config.integration_dir().join("fyre").join(&run_id); + if root.exists() { + return Err(AppFailure::from(anyhow::anyhow!( + "FYRE run {run_id} already exists at {}; use status or destroy with this run ID", + root.display() + ))); + } + fs::create_dir_all(root.join("results")) + .with_context(|| format!("failed to create FYRE run directory {}", root.display())) + .map_err(AppFailure::from)?; + copy_tree( + &self.config.asset_root().join("benchmarks/fyre/terraform"), + &root.join(TERRAFORM_DIRECTORY), + ) + .map_err(AppFailure::from)?; + let config_path = root.join("config.json"); + write_json(&config_path, &config).map_err(AppFailure::from)?; + let mut state = RunState { + schema_version: 1, + run_id: run_id.clone(), + phase: "initializing".to_owned(), + config_file: source, + current_scenario: None, + locust_helper_size: 0, + fast_time_helper_size: 0, + completed_scenarios: Vec::new(), + cleanup_required: true, + }; + write_state(&root, &state).map_err(AppFailure::from)?; + + let terraform = terraform_binary().map_err(AppFailure::from)?; + let init = self.fyre_environment( + CommandSpec::new(&terraform) + .args(["init", "-input=false"]) + .cwd(root.join(TERRAFORM_DIRECTORY)), + ); + let primary = async { + self.run_cancellable(&init).await?; + let validate = self.fyre_environment( + CommandSpec::new(&terraform) + .arg("validate") + .cwd(root.join(TERRAFORM_DIRECTORY)), + ); + self.run_cancellable(&validate).await?; + let matrix = tokio::time::timeout( + Duration::from_secs(config.workload.maximum_campaign_seconds.into()), + self.run_fyre_matrix( + &terraform, + &root, + &config_path, + &public_key, + &mut config, + &mut state, + ), + ) + .await; + match matrix { + Ok(result) => result?, + Err(_) => { + if let Some(scenario) = state.current_scenario.as_deref() { + let _ = self + .collect_fyre_scenario(&root, &config_path, scenario) + .await; + } + return Err(AppFailure::from(anyhow::anyhow!( + "FYRE campaign exceeded its configured time bound" + ))); + } + } + write_json( + &root.join("manifest.json"), + &json!({ + "schema_version": 1, + "run_id": run_id, + "configuration": config, + "state": state, + "terraform_lock": root.join(TERRAFORM_DIRECTORY).join(".terraform.lock.hcl"), + }), + ) + .map_err(AppFailure::from) + } + .await; + + state.phase = "collecting".to_owned(); + let _ = write_state(&root, &state); + let report_result = if primary.is_ok() { + self.generate_fyre_report(&root, &config_path).await + } else { + Ok(()) + }; + let primary = primary.and(report_result); + state.phase = "destroying".to_owned(); + let _ = write_state(&root, &state); + let cleanup = self.terraform_destroy(&terraform, &root).await; + if cleanup.is_ok() { + state.cleanup_required = false; + state.phase = if primary.is_ok() { + "complete" + } else { + "failed" + } + .to_owned(); + } else { + state.phase = "cleanup-failed".to_owned(); + } + let _ = write_state(&root, &state); + super::finish_with_cleanup(primary.err(), cleanup) + } + + #[allow(clippy::too_many_arguments)] + async fn run_fyre_matrix( + &self, + terraform: &OsString, + root: &Path, + config_path: &Path, + public_key: &Path, + config: &mut FyreConfig, + state: &mut RunState, + ) -> AppResult<()> { + let public_key = fs::read_to_string(public_key) + .context("failed to read FYRE SSH public key") + .map_err(AppFailure::from)?; + let mut index = 0; + while index < config.scenarios.len() { + let scenario = config.scenarios[index].clone(); + let locust = config.infrastructure.helper_sizes[state.locust_helper_size]; + let fast_time = config.infrastructure.helper_sizes[state.fast_time_helper_size]; + config.active_helper = Some(ActiveHelper { + locust_cpu: locust.cpu, + locust_memory_gb: locust.memory_gb, + fast_time_cpu: fast_time.cpu, + fast_time_memory_gb: fast_time.memory_gb, + }); + write_json(config_path, config).map_err(AppFailure::from)?; + state.current_scenario = Some(scenario.id.clone()); + state.phase = "provisioning".to_owned(); + write_state(root, state).map_err(AppFailure::from)?; + let variables = terraform_variables( + &state.run_id, + config, + &scenario, + &public_key, + self.fyre_text("FYRE_PRODUCT_GROUP_ID"), + self.fyre_text("FYRE_SITE"), + ); + write_json(&root.join(TERRAFORM_VARIABLES), &variables).map_err(AppFailure::from)?; + let apply = self.fyre_environment( + CommandSpec::new(terraform) + .args(["apply", "-input=false", "-auto-approve", "-var-file"]) + .arg(root.join(TERRAFORM_VARIABLES)) + .cwd(root.join(TERRAFORM_DIRECTORY)), + ); + self.run_cancellable(&apply).await?; + let inventory = self.terraform_inventory(terraform, root)?; + let scenario_root = root.join("results").join(&scenario.id); + if scenario_root.exists() { + fs::remove_dir_all(&scenario_root) + .with_context(|| format!("failed to reset {}", scenario_root.display())) + .map_err(AppFailure::from)?; + } + fs::create_dir_all(&scenario_root) + .with_context(|| format!("failed to create {}", scenario_root.display())) + .map_err(AppFailure::from)?; + let inventory_path = scenario_root.join("inventory.json"); + write_json(&inventory_path, &inventory).map_err(AppFailure::from)?; + state.phase = "benchmarking".to_owned(); + write_state(root, state).map_err(AppFailure::from)?; + let campaign = self.fyre_campaign_command(root, config_path, &scenario.id); + let campaign_result = self.run_cancellable(&campaign).await; + if campaign_result.is_err() { + let _ = self + .collect_fyre_scenario(root, config_path, &scenario.id) + .await; + } + match campaign_result { + Ok(()) => { + state.completed_scenarios.push(scenario.id); + write_state(root, state).map_err(AppFailure::from)?; + index += 1; + } + Err(AppFailure::Infrastructure( + crate::infrastructure::InfrastructureError::ChildExit { status, .. }, + )) if status.code() == Some(HELPER_SATURATION_EXIT) => { + let request: Value = serde_json::from_slice( + &fs::read(scenario_root.join("helper-request.json")) + .context("helper saturation did not produce helper-request.json") + .map_err(AppFailure::from)?, + ) + .context("invalid helper saturation request") + .map_err(AppFailure::from)?; + let role = request["role"].as_str().ok_or_else(|| { + AppFailure::from(anyhow::anyhow!( + "helper saturation request has an unknown role" + )) + })?; + let size = match role { + "locust" => &mut state.locust_helper_size, + "fast-time" => &mut state.fast_time_helper_size, + _ => { + return Err(AppFailure::from(anyhow::anyhow!( + "helper saturation request has an unknown role" + ))); + } + }; + *size += 1; + if *size >= config.infrastructure.helper_sizes.len() { + return Err(AppFailure::from(anyhow::anyhow!( + "helper headroom is inconclusive: the saturated helper reached the configured 16 vCPU / 32 GB limit" + ))); + } + let archive = root + .join("invalidated") + .join(format!("{role}-size-{}-at-{}", *size, scenario.id)); + fs::create_dir_all(&archive) + .with_context(|| format!("failed to create {}", archive.display())) + .map_err(AppFailure::from)?; + for scenario in &config.scenarios { + let path = root.join("results").join(&scenario.id); + if path.exists() { + let archived = archive.join(&scenario.id); + fs::rename(&path, &archived) + .with_context(|| { + format!( + "failed to archive {} as {}", + path.display(), + archived.display() + ) + }) + .map_err(AppFailure::from)?; + } + } + state.completed_scenarios.clear(); + index = 0; + } + Err(error) => return Err(error), + } + } + state.current_scenario = None; + Ok(()) + } + + fn fyre_campaign_command( + &self, + root: &Path, + config_path: &Path, + scenario: &str, + ) -> CommandSpec { + let scenario_root = root.join("results").join(scenario); + CommandSpec::new("python3") + .arg(self.config.asset_root().join("benchmarks/fyre/campaign.py")) + .arg("--config") + .arg(config_path) + .arg("--inventory") + .arg(scenario_root.join("inventory.json")) + .args(["--scenario", scenario]) + .arg("--deploy") + .arg(self.config.asset_root().join("benchmarks/fyre/deploy")) + .arg("--output") + .arg(scenario_root) + } + + async fn collect_fyre_scenario( + &self, + root: &Path, + config_path: &Path, + scenario: &str, + ) -> AppResult<()> { + let inventory = root.join("results").join(scenario).join("inventory.json"); + if !inventory.is_file() { + return Ok(()); + } + let collection = self + .fyre_campaign_command(root, config_path, scenario) + .arg("--collect-only"); + self.run_cancellable(&collection).await + } + + fn terraform_inventory(&self, terraform: &OsString, root: &Path) -> AppResult { + let command = self.fyre_environment( + CommandSpec::new(terraform) + .args(["output", "-json", "inventory"]) + .cwd(root.join(TERRAFORM_DIRECTORY)), + ); + let output = self + .runner + .capture_stdout(&command) + .map_err(AppFailure::from)?; + serde_json::from_slice(&output) + .context("Terraform inventory output is not valid JSON") + .map_err(AppFailure::from) + } + + async fn generate_fyre_report(&self, root: &Path, config: &Path) -> AppResult<()> { + let command = CommandSpec::new("uv") + .args(["run", "--with", "matplotlib==3.10.6"]) + .arg(self.config.asset_root().join("benchmarks/fyre/report.py")) + .arg("--config") + .arg(config) + .arg("--results") + .arg(root.join("results")); + self.run_cancellable(&command).await + } + + fn fyre_status(&self, run_id: &str) -> AppResult<()> { + validate_run_id(run_id).map_err(AppFailure::from)?; + let root = self.config.integration_dir().join("fyre").join(run_id); + let state = read_owned_state(&root, run_id).map_err(AppFailure::from)?; + println!( + "{}", + serde_json::to_string_pretty(&state) + .map_err(anyhow::Error::from) + .map_err(AppFailure::from)? + ); + Ok(()) + } + + async fn destroy_fyre(&self, run_id: &str) -> AppResult<()> { + validate_run_id(run_id).map_err(AppFailure::from)?; + let root = self.config.integration_dir().join("fyre").join(run_id); + let mut state = read_owned_state(&root, run_id).map_err(AppFailure::from)?; + let terraform = terraform_binary().map_err(AppFailure::from)?; + state.phase = "destroying".to_owned(); + write_state(&root, &state).map_err(AppFailure::from)?; + self.terraform_destroy(&terraform, &root).await?; + state.phase = "destroyed".to_owned(); + state.cleanup_required = false; + write_state(&root, &state).map_err(AppFailure::from) + } + + async fn terraform_destroy(&self, terraform: &OsString, root: &Path) -> AppResult<()> { + let variables = root.join(TERRAFORM_VARIABLES); + if !variables.is_file() { + return Ok(()); + } + let command = self.fyre_environment( + CommandSpec::new(terraform) + .args(["destroy", "-input=false", "-auto-approve", "-var-file"]) + .arg(variables) + .cwd(root.join(TERRAFORM_DIRECTORY)), + ); + let mut last = None; + for attempt in 0..3 { + match self.run_cancellable(&command).await { + Ok(()) => return Ok(()), + Err(error) => last = Some(error), + } + if attempt < 2 { + tokio::time::sleep(Duration::from_secs(2_u64.pow(attempt + 1))).await; + } + } + Err(last.unwrap_or_else(|| AppFailure::from(anyhow::anyhow!("Terraform destroy failed")))) + } + + async fn run_cancellable(&self, command: &CommandSpec) -> AppResult<()> { + let (sender, receiver) = tokio::sync::watch::channel(false); + let process = self.runner.run_async_cancellable(command, receiver); + tokio::pin!(process); + tokio::select! { + result = &mut process => result.map_err(AppFailure::from), + signal = tokio::signal::ctrl_c() => { + signal.context("failed to install interrupt handler").map_err(AppFailure::from)?; + sender.send_replace(true); + process.await.map_err(AppFailure::from) + } + } + } + + fn fyre_environment(&self, mut command: CommandSpec) -> CommandSpec { + for key in [ + "FYRE_USERNAME", + "FYRE_API_KEY", + "FYRE_PRODUCT_GROUP_ID", + "FYRE_SITE", + ] { + if let Some(value) = self + .config + .environment() + .get(OsStr::new(key)) + .map(|value| value.value.clone()) + { + command = command.env(key, value); + } + } + command + } + + fn fyre_text(&self, key: &str) -> Option<&str> { + self.config + .environment() + .get(OsStr::new(key)) + .and_then(|value| value.value.to_str()) + .filter(|value| !value.is_empty()) + } + + fn require_fyre_credentials(&self) -> AppResult<()> { + for key in ["FYRE_USERNAME", "FYRE_API_KEY"] { + if self.fyre_text(key).is_none() { + return Err(AppFailure::from(anyhow::anyhow!( + "{key} is required for FYRE provisioning" + ))); + } + } + Ok(()) + } +} + +fn terraform_variables( + run_id: &str, + config: &FyreConfig, + scenario: &Scenario, + public_key: &str, + product_group_id: Option<&str>, + site: Option<&str>, +) -> Value { + let helpers = config + .active_helper + .as_ref() + .expect("active helper must be set"); + json!({ + "run_id": run_id, + "os": config.infrastructure.os, + "ssh_public_key": public_key.trim(), + "expiry_hours": config.infrastructure.expiry_hours, + "dataplane_count": scenario.replicas, + "dataplane_cpu": scenario.cpu, + "dataplane_memory_gb": scenario.memory_gb, + "locust_cpu": helpers.locust_cpu, + "locust_memory_gb": helpers.locust_memory_gb, + "fast_time_cpu": helpers.fast_time_cpu, + "fast_time_memory_gb": helpers.fast_time_memory_gb, + "product_group_id": product_group_id, + "site": site, + }) +} + +fn read_config(path: &Path) -> Result { + let source = fs::read(path) + .with_context(|| format!("failed to read FYRE configuration {}", path.display()))?; + yaml_serde::from_slice(&source) + .with_context(|| format!("failed to parse FYRE configuration {}", path.display())) +} + +fn validate_config(config: &FyreConfig) -> Result<()> { + ensure!( + config.schema_version == 1, + "unsupported FYRE configuration schema" + ); + ensure!( + config.infrastructure.os == "Ubuntu 24.04", + "FYRE benchmark OS must be Ubuntu 24.04" + ); + ensure!( + config.infrastructure.expiry_hours == 8, + "FYRE expiry must remain eight hours" + ); + ensure!( + config.workload.protocol_version == "2026-07-28", + "FYRE load supports only modern 2026-07-28" + ); + ensure!( + config.workload.first_users >= 125, + "FYRE load must start at 125 users or more" + ); + ensure!( + config.workload.maximum_users <= 32_000, + "FYRE load must be bounded at 32,000 users" + ); + ensure!( + config.workload.maximum_campaign_seconds <= 21_600, + "FYRE campaign must be bounded at six hours" + ); + ensure!( + config.workload.repetitions == 3, + "candidate capacity must use three repetitions" + ); + let expected_tools = BTreeSet::from([ + "convert_time", + "echo", + "get_stats", + "get_system_time", + "schema_success", + "verify-protocol", + ]); + let actual_tools = config + .workload + .tools + .iter() + .map(String::as_str) + .collect::>(); + ensure!( + actual_tools == expected_tools, + "FYRE workload must contain the six nonfailure Fast Time tools" + ); + ensure!( + !config.infrastructure.helper_sizes.is_empty(), + "at least one helper size is required" + ); + let maximum = config + .infrastructure + .helper_sizes + .last() + .expect("nonempty helper sizes"); + ensure!( + maximum.cpu <= 16 && maximum.memory_gb <= 32, + "helper resources exceed 16 vCPU / 32 GB" + ); + let mut ids = BTreeSet::<&str>::new(); + for scenario in &config.scenarios { + validate_run_id(&scenario.id)?; + ensure!( + ids.insert(scenario.id.as_str()), + "duplicate FYRE scenario {}", + scenario.id + ); + ensure!( + scenario.replicas > 0 && scenario.cpu > 0 && scenario.memory_gb > 0, + "scenario {} has zero resources", + scenario.id + ); + ensure!( + scenario.replicas * scenario.cpu == scenario.multiplier * 2, + "scenario {} CPU total does not match its multiplier", + scenario.id + ); + ensure!( + scenario.replicas * scenario.memory_gb == scenario.multiplier * 8, + "scenario {} memory total does not match its multiplier", + scenario.id + ); + } + ensure!(ids.contains("baseline"), "FYRE matrix requires baseline"); + for image in [ + &config.images.dataplane, + &config.images.fast_time, + &config.images.helpers, + &config.images.locust, + &config.images.redis, + ] { + ensure!( + image.contains("@sha256:"), + "all benchmark images must be pinned by digest" + ); + } + Ok(()) +} + +fn validate_run_id(run_id: &str) -> Result<()> { + ensure!( + !run_id.is_empty() + && run_id.len() <= 48 + && run_id + .bytes() + .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || byte == b'-') + && run_id + .as_bytes() + .first() + .is_some_and(u8::is_ascii_alphanumeric) + && run_id + .as_bytes() + .last() + .is_some_and(u8::is_ascii_alphanumeric), + "run ID must contain only lowercase letters, digits, and internal hyphens" + ); + Ok(()) +} + +fn expand_home(path: &Path) -> Result { + let text = path.to_str().context("SSH key path must be UTF-8")?; + if text == "~" || text.starts_with("~/") { + let home = std::env::var_os("HOME").context("HOME is required to expand SSH key paths")?; + return Ok(PathBuf::from(home).join(text.trim_start_matches("~/"))); + } + Ok(path.to_path_buf()) +} + +fn ensure_file(path: &Path, label: &str) -> Result<()> { + ensure!(path.is_file(), "{label} {} does not exist", path.display()); + Ok(()) +} + +fn terraform_binary() -> Result { + if let Some(binary) = std::env::var_os("CF_TERRAFORM_BIN") { + ensure!(!binary.is_empty(), "CF_TERRAFORM_BIN must not be empty"); + return Ok(binary); + } + if std::process::Command::new("terraform") + .arg("version") + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::null()) + .status() + .is_ok_and(|status| status.success()) + { + return Ok(OsString::from("terraform")); + } + bail!("Terraform is required; set CF_TERRAFORM_BIN to its executable") +} + +fn copy_tree(source: &Path, destination: &Path) -> Result<()> { + fs::create_dir_all(destination) + .with_context(|| format!("failed to create {}", destination.display()))?; + for entry in + fs::read_dir(source).with_context(|| format!("failed to read {}", source.display()))? + { + let entry = entry?; + let target = destination.join(entry.file_name()); + if entry.file_type()?.is_dir() { + copy_tree(&entry.path(), &target)?; + } else { + fs::copy(entry.path(), &target) + .with_context(|| format!("failed to copy {}", entry.path().display()))?; + } + } + Ok(()) +} + +fn write_json(path: &Path, value: &impl Serialize) -> Result<()> { + let temporary = path.with_extension("tmp"); + fs::write(&temporary, serde_json::to_vec_pretty(value)?) + .with_context(|| format!("failed to write {}", temporary.display()))?; + fs::rename(&temporary, path).with_context(|| format!("failed to activate {}", path.display())) +} + +fn write_state(root: &Path, state: &RunState) -> Result<()> { + write_json(&root.join(OWNERSHIP_FILE), state) +} + +fn read_owned_state(root: &Path, expected_run_id: &str) -> Result { + let path = root.join(OWNERSHIP_FILE); + let state: RunState = serde_json::from_slice( + &fs::read(&path) + .with_context(|| format!("FYRE run state {} does not exist", path.display()))?, + ) + .context("invalid FYRE run state")?; + ensure!( + state.run_id == expected_run_id, + "FYRE run ownership mismatch; refusing cleanup" + ); + ensure!( + root.join(TERRAFORM_DIRECTORY).is_dir(), + "FYRE Terraform state directory is missing; refusing cleanup" + ); + Ok(state) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn packaged_matrix_is_valid_and_matched() { + let config = read_config( + Path::new(env!("CARGO_MANIFEST_DIR")) + .join("benchmarks/fyre/scaling.yaml") + .as_path(), + ) + .expect("packaged FYRE config"); + validate_config(&config).expect("valid FYRE config"); + assert_eq!(config.scenarios.len(), 6); + } + + #[test] + fn cleanup_requires_matching_owned_state() { + let directory = tempfile::tempdir().expect("temporary directory"); + fs::create_dir(directory.path().join(TERRAFORM_DIRECTORY)).expect("terraform directory"); + let state = RunState { + schema_version: 1, + run_id: "owned-run".to_owned(), + phase: "failed".to_owned(), + config_file: PathBuf::from("config.yaml"), + current_scenario: None, + locust_helper_size: 0, + fast_time_helper_size: 0, + completed_scenarios: Vec::new(), + cleanup_required: true, + }; + write_state(directory.path(), &state).expect("state"); + let error = + read_owned_state(directory.path(), "another-run").expect_err("ownership mismatch"); + assert!(error.to_string().contains("ownership mismatch")); + } + + #[test] + fn run_ids_reject_paths_and_uppercase() { + for invalid in ["../manual-vm", "UPPER", "-leading", "trailing-"] { + assert!(validate_run_id(invalid).is_err(), "{invalid}"); + } + } +} diff --git a/src/runtime/mod.rs b/src/runtime/mod.rs index 39509b1..5236c0b 100644 --- a/src/runtime/mod.rs +++ b/src/runtime/mod.rs @@ -65,6 +65,7 @@ const LOAD_TARGET_CPUSET_ENV: &str = "CF_LOAD_TARGET_CPUSET"; mod ci; mod conformance; mod control_plane; +mod fyre; mod inspect; mod live; mod performance; @@ -110,6 +111,7 @@ impl RuntimeContext { .await } Action::Load(args) => self.run_load(args).await, + Action::Fyre(action) => self.execute_fyre(action).await, Action::Live { lane, group,