diff --git a/CHANGELOG.md b/CHANGELOG.md index 653f226..0791633 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,27 @@ and this project uses [Semantic Versioning](https://semver.org/spec/v2.0.0.html) ## [Unreleased] +### Added + +- Add a FYRE OpenShift comparison profile that provisions 40 GB nodes and runs + the built-in and external dataplane lanes concurrently on separate target, + Locust, and Fast Time workers. Package the pinned `oc` runtime, collect + gateway-side memory, and produce a self-contained Markdown report alongside + the existing CSV, JSON, and Slack-ready PNG artifacts. +- Add a quota-sized 2 vCPU / 2 GiB OpenShift profile that runs all eight + built-in/external and 125/250/500/1,000-user measurements concurrently with + exact pod limits on three dedicated-role workers. Reserve 1.75 vCPU / 1.75 + GiB per 2 vCPU / 2 GiB target so all eight targets schedule with host + headroom, set 40 GB master and worker root disks, disable FYRE's automatic + worker data disks, and preserve structured FYRE API validation errors. + +### Changed + +- Generalize `load fyre run|status|destroy` ownership and cleanup to support + run-owned OpenShift clusters as well as standalone Terraform VMs. +- Resume an interrupted OpenShift campaign when `load fyre run` is invoked + again with the same run ID and saved configuration. + ## [0.5.0] - 2026-09-16 ### Added diff --git a/benchmarks/fyre/README.md b/benchmarks/fyre/README.md index c14bc32..3682b43 100644 --- a/benchmarks/fyre/README.md +++ b/benchmarks/fyre/README.md @@ -44,6 +44,11 @@ disk and expose no create-time root-disk setting. The default three-VM run therefore needs 750 GB of FYRE disk quota. The CLI checks CPU, memory, disk, and public-IP quota before it creates any benchmark VM. +The OpenShift profile requires `FYRE_PRODUCT_GROUP_ID`, Docker on the +orchestration host, and access to the FYRE OpenShift API and cluster DNS. The +CLI runs a digest-pinned OpenShift client container, so a host `oc` installation +is not required. + ## Run the complete comparison The bare command is the CI entrypoint for the complete eight-run comparison: @@ -137,3 +142,98 @@ cf-integration load fyre run \ Comparison reports derive the target allocation from the selected profile; both lanes always run sequentially on that same VM. + +## Parallel OpenShift profile with 40 GB disks + +`openshift.yaml` runs the same eight comparison measurements on a FYRE +OpenShift cluster while reducing every master and worker root disk to 40 GB +and explicitly disabling FYRE's default worker data disks. +The built-in and external lanes run concurrently and remain isolated on six +dedicated workers: + +| Lane role | Workers | Pod allocation on each worker | +| --- | ---: | ---: | +| Built-in and external Locust | 2 | 4 vCPU / 16 GiB each | +| Built-in and external target | 2 | 4 vCPU / 4 GiB each | +| Built-in and external Fast Time | 2 | 8 vCPU / 32 GiB each | + +Each Locust pod contains one master and three workers. Each target allocation +includes its supporting PostgreSQL/Redis or Redis/JWKS containers. Each lane +has a separate zero-delay Fast Time service. No measured target, load generator, +or backend shares a worker node with the other lane. + +Run the complete parallel comparison with: + +```bash +cf-integration load fyre run \ + --file benchmarks/fyre/openshift.yaml \ + --run-id openshift-builtin-external +``` + +The short form is: + +```bash +cf-integration l f r -f benchmarks/fyre/openshift.yaml -i openshift-builtin-external +``` + +The command creates the cluster through the FYRE OpenShift API, authenticates +with the generated kubeadmin credential, assigns the six workers by their +configured CPU and memory, runs both lanes in parallel at 125, 250, 500, and +1,000 users, downloads every phase before deleting its pods, writes the final +report, deletes the benchmark namespace, and deletes only the run-owned +cluster. A failed or interrupted campaign retains its local run state and +retries cluster cleanup three times. Run the same command with the same run ID +to resume an interrupted OpenShift campaign from its saved manifest. + +OpenShift artifacts use the same +`$CF_INTEGRATION_DIR/fyre//results/` layout and add `report.md`, a +self-contained report with the result table, memory averages and peaks, +request mix, pinned images, and Mermaid architecture. The cluster record and +manifest recursively omit passwords, tokens, pull secrets, API keys, and +kubeconfig data. + +The built-in lane remains pinned to the MCP SDK v2 fixture image until that SDK +change is available in the main gateway image. The profile must not be changed +back to the main image before that merge because both lanes use the same modern +`2026-07-28` client. + +### Fully parallel 2 vCPU / 2 GiB comparison + +`openshift-2v2-parallel.yaml` runs all eight measurements at the same time: +built-in and external dataplane lanes at 125, 250, 500, and 1,000 users. Each +measurement gets its own target, Locust, and Fast Time pods with identical +requests and limits. Pods share only with pods serving the same role. + +| Dedicated worker role | Worker size | Pods | Reserved per measurement | +| --- | ---: | ---: | ---: | +| Target | 16 vCPU / 16 GiB | 8 | 2 vCPU / 2 GiB limits; 1.75 vCPU / 1.75 GiB requests | +| Locust | 14 vCPU / 12 GiB | 8 | 1.5 vCPU / 1.25 GiB | +| Fast Time | 14 vCPU / 12 GiB | 8 | 1.5 vCPU / 1.375 GiB | + +Each target reservation includes its supporting PostgreSQL and Redis +containers for the built-in dataplane, or Redis and loopback JWKS containers +for the external dataplane. Each Locust pod has one master and three workers. +The helper pressure gate rejects the campaign if the shared helper workers or +individual helper pods become the bottleneck. + +Run the full comparison with one command: + +```bash +cf-integration load fyre run \ + --file benchmarks/fyre/openshift-2v2-parallel.yaml \ + --run-id openshift-2v2-parallel +``` + +The short form is: + +```bash +cf-integration l f r -f benchmarks/fyre/openshift-2v2-parallel.yaml -i openshift-2v2-parallel +``` + +The three OpenShift masters and all three workers use 40 GB root disks, the +FYRE API VM uses its fixed 500 GB root disk, and workers have no additional +data disks. The cluster therefore requests 740 GB of disk in total. The +orchestration command may run on a persistent VM or CI worker; the benchmark +continues if the developer laptop sleeps. Artifacts are downloaded to +`$CF_INTEGRATION_DIR/fyre//results/` before the run-owned cluster is +deleted. diff --git a/benchmarks/fyre/openshift-2v2-parallel.yaml b/benchmarks/fyre/openshift-2v2-parallel.yaml new file mode 100644 index 0000000..12b936c --- /dev/null +++ b/benchmarks/fyre/openshift-2v2-parallel.yaml @@ -0,0 +1,54 @@ +schema_version: 1 +benchmark_kind: comparison +infrastructure: + kind: openshift + os: Red Hat CoreOS + expiry_hours: 24 + openshift: + version: "4.21.27" + base_disk_gb: 40 + oc_image: quay.io/openshift/origin-cli@sha256:f3fc63fc3d2fd1c3a2c3311bf517b776f6596244e697dad32d62ca9fd6482fe5 + master: { cpu: 4, memory_gb: 16 } + api: { cpu: 4, memory_gb: 8 } + target_pod: { cpu_millicores: 1750, memory_mib: 1792 } + load_pod: { cpu_millicores: 1500, memory_mib: 1280 } + backend_pod: { cpu_millicores: 1500, memory_mib: 1408 } + worker_pools: + - { role: target, cpu: 16, memory_gb: 16, count: 1 } + - { role: locust, cpu: 14, memory_gb: 12, count: 1 } + - { role: fast-time, cpu: 14, memory_gb: 12, count: 1 } +images: + dataplane: ghcr.io/contextforge-org/contextforge-data-plane@sha256:0b3026b21659ce6c494142aee2c5866b44122232e73fcf21102da1f210692a96 + controlplane: ghcr.io/contextforge-org/cf-integration-fixture@sha256:5b206e6f863cea9f8cabea6451392428fbe67b16bdf10fefd1f7ac8032d95b52 + fast_time: ghcr.io/ibm/cfex-mcp-fast-time-server@sha256:110e1826f5d763e5afadba770b731dac93e0819c1bbadb68671b0124260603cf + helpers: ghcr.io/contextforge-org/cf-integration-helpers@sha256:f7e557e263737328225d827e73b519d0b40facb2698c55ee3d2de0ed82f00fa3 + locust: mirror.gcr.io/locustio/locust@sha256:fd39232c31971fe7509582717a95592f0fbf9a917b705f355bd7471e03bdd649 + postgres: postgres@sha256:4ef4dbc939d61acea57712655ddb4b4ab27419c913f94cca0cd57cb3ea3c2280 + redis: mirror.gcr.io/library/redis@sha256:a7859ed111db3c1f5404a973a4747505d559fb5ca32d37e447afc0ef845a2103 +workload: + protocol_version: 2026-07-28 + user_levels: [125, 250, 500, 1000] + first_users: 125 + maximum_users: 1000 + ramp_seconds: 30 + warmup_seconds: 30 + measure_seconds: 3600 + repetitions: 1 + maximum_campaign_seconds: 7200 + 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 + parallel_lanes: true + parallel_user_levels: true + tools: + - convert_time + - echo + - get_stats + - get_system_time + - schema_success + - verify-protocol +scenarios: + - { id: comparison, label: Built-in dataplane vs external dataplane, replicas: 1, cpu: 2, memory_gb: 2, multiplier: 1 } diff --git a/benchmarks/fyre/openshift.yaml b/benchmarks/fyre/openshift.yaml new file mode 100644 index 0000000..d14a98c --- /dev/null +++ b/benchmarks/fyre/openshift.yaml @@ -0,0 +1,57 @@ +schema_version: 1 +benchmark_kind: comparison +infrastructure: + kind: openshift + os: Red Hat CoreOS + expiry_hours: 12 + openshift: + version: "4.21.27" + base_disk_gb: 40 + oc_image: quay.io/openshift/origin-cli@sha256:f3fc63fc3d2fd1c3a2c3311bf517b776f6596244e697dad32d62ca9fd6482fe5 + master: { cpu: 4, memory_gb: 16 } + api: { cpu: 4, memory_gb: 8 } + target_pod: { cpu_millicores: 3500, memory_mib: 3584 } + load_pod: { cpu_millicores: 4000, memory_mib: 16384 } + backend_pod: { cpu_millicores: 8000, memory_mib: 32768 } + worker_pools: + - { role: target-builtin, cpu: 6, memory_gb: 12, count: 1 } + - { role: target-external, cpu: 6, memory_gb: 12, count: 1 } + - { role: locust-builtin, cpu: 6, memory_gb: 20, count: 1 } + - { role: locust-external, cpu: 6, memory_gb: 20, count: 1 } + - { role: fast-time-builtin, cpu: 10, memory_gb: 40, count: 1 } + - { role: fast-time-external, cpu: 10, memory_gb: 40, count: 1 } +images: + dataplane: ghcr.io/contextforge-org/contextforge-data-plane@sha256:0b3026b21659ce6c494142aee2c5866b44122232e73fcf21102da1f210692a96 + controlplane: ghcr.io/contextforge-org/cf-integration-fixture@sha256:5b206e6f863cea9f8cabea6451392428fbe67b16bdf10fefd1f7ac8032d95b52 + fast_time: ghcr.io/ibm/cfex-mcp-fast-time-server@sha256:110e1826f5d763e5afadba770b731dac93e0819c1bbadb68671b0124260603cf + helpers: ghcr.io/contextforge-org/cf-integration-helpers@sha256:f7e557e263737328225d827e73b519d0b40facb2698c55ee3d2de0ed82f00fa3 + locust: mirror.gcr.io/locustio/locust@sha256:fd39232c31971fe7509582717a95592f0fbf9a917b705f355bd7471e03bdd649 + postgres: postgres@sha256:4ef4dbc939d61acea57712655ddb4b4ab27419c913f94cca0cd57cb3ea3c2280 + redis: mirror.gcr.io/library/redis@sha256:a7859ed111db3c1f5404a973a4747505d559fb5ca32d37e447afc0ef845a2103 +workload: + protocol_version: 2026-07-28 + user_levels: [125, 250, 500, 1000] + first_users: 125 + maximum_users: 1000 + ramp_seconds: 30 + warmup_seconds: 30 + measure_seconds: 3600 + repetitions: 1 + maximum_campaign_seconds: 18000 + 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 + parallel_lanes: true + parallel_user_levels: false + tools: + - convert_time + - echo + - get_stats + - get_system_time + - schema_success + - verify-protocol +scenarios: + - { id: comparison, label: Built-in dataplane vs external dataplane, replicas: 1, cpu: 4, memory_gb: 4, multiplier: 1 } diff --git a/benchmarks/fyre/openshift_campaign.py b/benchmarks/fyre/openshift_campaign.py new file mode 100644 index 0000000..e68c4e0 --- /dev/null +++ b/benchmarks/fyre/openshift_campaign.py @@ -0,0 +1,1494 @@ +#!/usr/bin/env python3 +"""Run the built-in and external FYRE OpenShift lanes in parallel.""" + +from __future__ import annotations + +import argparse +import concurrent.futures +import json +import secrets +import subprocess +import threading +import time +from pathlib import Path + +from campaign import read_stats + +LANES = ("builtin", "external") +SERVER_ID = "fyre-fast-time" + + +class Oc: + def __init__(self, image: str, kubeconfig: Path): + self.image = image + self.kubeconfig = kubeconfig.resolve() + self.mount = self.kubeconfig.parent + + def run( + self, + *arguments: str, + input_text: str | None = None, + check: bool = True, + timeout: int = 300, + ) -> subprocess.CompletedProcess[str]: + command = [ + "docker", + "run", + "--rm", + "-i", + "--platform", + "linux/amd64", + "-v", + f"{self.mount}:/work", + "-e", + "KUBECONFIG=/work/kubeconfig", + self.image, + "oc", + *arguments, + ] + result = subprocess.run( + command, + input=input_text, + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + check=False, + timeout=timeout, + ) + if check and result.returncode: + raise RuntimeError( + f"oc command exited {result.returncode}: " + f"{result.stderr.strip() or 'no error output'}" + ) + return result + + def container_path(self, path: Path) -> str: + relative = path.resolve().relative_to(self.mount) + return f"/work/{relative.as_posix()}" + + def json(self, *arguments: str) -> dict: + return json.loads(self.run(*arguments, "-o", "json").stdout) + + def apply(self, resource: dict) -> None: + self.run("apply", "-f", "-", input_text=json.dumps(resource)) + + def delete(self, *arguments: str) -> None: + self.run("delete", *arguments, "--ignore-not-found", check=False) + + +def metadata(name: str, namespace: str | None = None) -> dict: + result = {"name": name, "labels": {"app.kubernetes.io/part-of": "cf-fyre"}} + if namespace: + result["namespace"] = namespace + return result + + +def resources( + cpu: str, + memory: str, + request_cpu: str | None = None, + request_memory: str | None = None, +) -> dict: + return { + "requests": { + "cpu": request_cpu or cpu, + "memory": request_memory or memory, + }, + "limits": {"cpu": cpu, "memory": memory}, + } + + +def pod_size(config: dict, key: str) -> tuple[str, str]: + size = config["infrastructure"]["openshift"][key] + return f"{int(size['cpu_millicores'])}m", f"{int(size['memory_mib'])}Mi" + + +def target_size(config: dict) -> tuple[int, int]: + scenario = config["scenarios"][0] + return int(scenario["cpu"]) * 1000, int(scenario["memory_gb"]) * 1024 + + +def target_request_size(config: dict) -> tuple[int, int]: + size = config["infrastructure"]["openshift"]["target_pod"] + return int(size["cpu_millicores"]), int(size["memory_mib"]) + + +def instance_name(base: str, lane: str, users: int | None = None) -> str: + suffix = f"-{users}" if users is not None else "" + return f"{base}-{lane}{suffix}" + + +def with_users(base: str, users: int | None) -> str: + return f"{base}-{users}" if users is not None else base + + +def env_list(values: dict[str, str]) -> list[dict]: + return [{"name": key, "value": value} for key, value in values.items()] + + +def secret_env(name: str) -> dict: + return { + "name": "MCPGATEWAY_BEARER_TOKEN", + "valueFrom": {"secretKeyRef": {"name": name, "key": "token"}}, + } + + +def wait_for( + oc: Oc, + namespace: str, + kind: str, + name: str, + condition: str, + timeout_seconds: int, +) -> None: + wait_expression = ( + "--for=jsonpath={.status.phase}=Succeeded" + if condition == "phase=Succeeded" + else f"--for={condition}" + ) + result = oc.run( + "wait", + wait_expression, + f"{kind}/{name}", + "-n", + namespace, + f"--timeout={timeout_seconds}s", + check=False, + timeout=timeout_seconds + 30, + ) + if result.returncode: + describe = oc.run( + "describe", f"{kind}/{name}", "-n", namespace, check=False + ).stdout + raise RuntimeError( + f"{kind}/{name} did not reach {condition}: {result.stderr}\n{describe}" + ) + + +def memory_gib(value: str) -> float: + suffixes = {"Ki": 1 / 1024 / 1024, "Mi": 1 / 1024, "Gi": 1.0} + for suffix, multiplier in suffixes.items(): + if value.endswith(suffix): + return float(value[: -len(suffix)]) * multiplier + return float(value) / 1024 / 1024 / 1024 + + +def assign_nodes(config: dict, nodes: dict) -> dict[str, list[str]]: + workers = [] + for item in nodes.get("items", []): + labels = item.get("metadata", {}).get("labels", {}) + if any( + key in labels + for key in ( + "node-role.kubernetes.io/master", + "node-role.kubernetes.io/control-plane", + ) + ): + continue + capacity = item.get("status", {}).get("capacity", {}) + workers.append( + { + "name": item["metadata"]["name"], + "cpu": int(capacity.get("cpu", 0)), + "memory_gb": memory_gib(capacity.get("memory", "0")), + } + ) + pools = config["infrastructure"]["openshift"]["worker_pools"] + if len(workers) != sum(int(pool["count"]) for pool in pools): + raise RuntimeError( + f"expected {sum(int(pool['count']) for pool in pools)} dedicated workers, " + f"found {len(workers)}: " + + ", ".join(item["name"] for item in workers) + ) + remaining = list(workers) + assigned: dict[str, list[str]] = {} + for pool in sorted(pools, key=lambda item: (item["memory_gb"], item["role"])): + assigned[pool["role"]] = [] + for _index in range(int(pool["count"])): + matches = sorted( + remaining, + key=lambda node: ( + abs(node["memory_gb"] - float(pool["memory_gb"])), + abs(node["cpu"] - int(pool["cpu"])), + node["name"], + ), + ) + if not matches: + raise RuntimeError(f"no OpenShift worker remains for {pool['role']}") + selected = matches[0] + if selected["cpu"] != int(pool["cpu"]) or abs( + selected["memory_gb"] - float(pool["memory_gb"]) + ) > 2: + raise RuntimeError( + f"worker {selected['name']} does not match {pool['role']} " + f"({selected['cpu']} vCPU / {selected['memory_gb']:.1f} GiB)" + ) + assigned[pool["role"]].append(selected["name"]) + remaining.remove(selected) + return assigned + + +def assigned_node( + nodes: dict[str, list[str]], role: str, lane: str, users: int | None +) -> str: + candidates = nodes.get(f"{role}-{lane}", nodes.get(role, [])) + if not candidates: + raise RuntimeError(f"no OpenShift worker assigned for {role}-{lane}") + lane_offset = 0 if lane == "builtin" else 1 + return candidates[((users or 0) + lane_offset) % len(candidates)] + + +def setup_namespace(oc: Oc, namespace: str, assets: Path) -> None: + oc.apply( + { + "apiVersion": "v1", + "kind": "Namespace", + "metadata": metadata(namespace), + } + ) + oc.apply( + { + "apiVersion": "v1", + "kind": "ServiceAccount", + "metadata": metadata("benchmark", namespace), + } + ) + oc.run( + "adm", + "policy", + "add-scc-to-user", + "anyuid", + "-z", + "benchmark", + "-n", + namespace, + ) + files = { + "locustfile_mcp.py": assets.parent.parent.joinpath( + "scripts/locustfile_mcp.py" + ).read_text(encoding="utf-8"), + "smoke.py": assets.joinpath("deploy/smoke.py").read_text(encoding="utf-8"), + "register_builtin.py": assets.joinpath( + "deploy/register_builtin.py" + ).read_text(encoding="utf-8"), + } + oc.apply( + { + "apiVersion": "v1", + "kind": "ConfigMap", + "metadata": metadata("benchmark-code", namespace), + "data": files, + } + ) + + +def service(oc: Oc, namespace: str, name: str, selector: str, ports: list[dict]) -> None: + oc.apply( + { + "apiVersion": "v1", + "kind": "Service", + "metadata": metadata(name, namespace), + "spec": { + "selector": {"cf.contextforge/service": selector}, + "ports": ports, + }, + } + ) + + +def deploy_fast_time( + oc: Oc, + config: dict, + namespace: str, + lane: str, + node: str, + users: int | None = None, +) -> None: + name = instance_name("fast-time", lane, users) + cpu, memory = pod_size(config, "backend_pod") + oc.apply( + { + "apiVersion": "v1", + "kind": "Pod", + "metadata": { + **metadata(name, namespace), + "labels": {"cf.contextforge/service": name}, + }, + "spec": { + "serviceAccountName": "benchmark", + "nodeName": node, + "restartPolicy": "Always", + "containers": [ + { + "name": "fast-time", + "image": config["images"]["fast_time"], + "env": env_list( + {"BIND_ADDRESS": "0.0.0.0:9080", "RUST_LOG": "warn"} + ), + "ports": [{"containerPort": 9080}], + "resources": resources(cpu, memory), + "readinessProbe": { + "httpGet": {"path": "/health", "port": 9080}, + "periodSeconds": 2, + "failureThreshold": 90, + }, + } + ], + }, + } + ) + service( + oc, + namespace, + name, + name, + [{"name": "http", "port": 9080, "targetPort": 9080}], + ) + wait_for(oc, namespace, "pod", name, "condition=Ready", 300) + + +def deploy_external( + oc: Oc, + config: dict, + namespace: str, + node: str, + users: int | None = None, +) -> tuple[str, list[str]]: + name = instance_name("target", "external", users) + total_cpu, total_memory = target_size(config) + request_cpu, request_memory = target_request_size(config) + dataplane_cpu = total_cpu - 500 + dataplane_memory = total_memory - 512 + dataplane_request_cpu = request_cpu - 500 + dataplane_request_memory = request_memory - 512 + oc.apply( + { + "apiVersion": "v1", + "kind": "Pod", + "metadata": { + **metadata(name, namespace), + "labels": {"cf.contextforge/service": name}, + }, + "spec": { + "serviceAccountName": "benchmark", + "nodeName": node, + "restartPolicy": "Always", + "volumes": [{"name": "keys", "emptyDir": {}}], + "containers": [ + { + "name": "redis", + "image": config["images"]["redis"], + "args": ["redis-server", "--save", "", "--appendonly", "no"], + "resources": resources("250m", "256Mi"), + "readinessProbe": { + "exec": {"command": ["redis-cli", "ping"]}, + "periodSeconds": 2, + }, + }, + { + "name": "dataplane", + "image": config["images"]["dataplane"], + "env": env_list( + { + "CONTEXTFORGE_DATA_PLANE_ADDRESS": "0.0.0.0:4445", + "CONTEXTFORGE_DATA_PLANE_REDIS_HOSTNAME": "127.0.0.1", + "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": f"{name}:4445,127.0.0.1:4445,localhost:4445", + "CONTEXTFORGE_DATA_PLANE_USER_CONFIG_CACHE_EXPIRY_SECONDS": str( + config["workload"]["config_cache_seconds"] + ), + "RUST_LOG": "warn", + } + ), + "ports": [{"containerPort": 4445}], + "resources": resources( + f"{dataplane_cpu}m", + f"{dataplane_memory}Mi", + f"{dataplane_request_cpu}m", + f"{dataplane_request_memory}Mi", + ), + "readinessProbe": { + "tcpSocket": {"port": 4445}, + "periodSeconds": 2, + "failureThreshold": 90, + }, + }, + { + "name": "auth", + "image": config["images"]["helpers"], + "args": ["__helper", "auth"], + "volumeMounts": [{"name": "keys", "mountPath": "/keys"}], + "resources": resources("250m", "256Mi"), + "readinessProbe": { + "exec": { + "command": ["cf-integration", "__helper", "health"] + }, + "periodSeconds": 2, + "failureThreshold": 90, + }, + }, + ], + }, + } + ) + service( + oc, + namespace, + name, + name, + [{"name": "mcp", "port": 4445, "targetPort": 4445}], + ) + wait_for(oc, namespace, "pod", name, "condition=Ready", 300) + token = oc.run( + "exec", + name, + "-n", + namespace, + "-c", + "auth", + "--", + "cf-integration", + "__helper", + "token", + "fyre-benchmark", + "fyre-user", + ).stdout.strip() + if not token or "\n" in token: + raise RuntimeError("external dataplane token helper returned invalid output") + backend = f"http://{instance_name('fast-time', 'external', users)}:9080/mcp" + tools = configure_external( + oc, + name, + namespace, + token, + backend, + config["workload"]["protocol_version"], + ) + return token, tools + + +def configure_external( + oc: Oc, + pod: str, + namespace: str, + token: str, + backend: str, + protocol_version: str, +) -> list[str]: + result = oc.run( + "exec", + "-i", + pod, + "-n", + namespace, + "-c", + "auth", + "--", + "/bin/sh", + "-ceu", + "IFS= read -r MCP_CONFORMANCE_TOKEN; export MCP_CONFORMANCE_TOKEN; export CF_CONFIG_REDIS_URL=redis://127.0.0.1:6379; exec cf-integration __helper fixture \"$@\"", + "fixture-config", + SERVER_ID, + backend, + protocol_version, + input_text=f"{token}\n", + ).stdout.splitlines() + if not result: + raise RuntimeError("external dataplane configuration produced no result") + return json.loads(result[-1]) + + +def builtin_environment( + secret: dict[str, str], database_host: str = "builtin-db" +) -> dict[str, str]: + return { + "DATABASE_URL": f"postgresql+psycopg://postgres:{secret['postgres']}@{database_host}:5432/mcp", + "REDIS_URL": f"redis://{database_host}:6379/0", + "CACHE_TYPE": "redis", + "JWT_ALGORITHM": "HS256", + "JWT_SECRET_KEY": secret["jwt"], + "JWT_AUDIENCE": "mcpgateway-api", + "JWT_ISSUER": "mcpgateway", + "AUTH_ENCRYPTION_SECRET": secret["encryption"], + "DEFAULT_USER_PASSWORD": secret["password"], + "PLATFORM_ADMIN_EMAIL": "admin@example.com", + "PLATFORM_ADMIN_PASSWORD": secret["password"], + "PASSWORD_CHANGE_ENFORCEMENT_ENABLED": "false", + "AUTH_REQUIRED": "true", + "MCP_CLIENT_AUTH_ENABLED": "true", + "MCP_REQUIRE_AUTH": "true", + "REQUIRE_USER_IN_DB": "false", + "LOG_LEVEL": "WARNING", + "TRANSPORT_TYPE": "streamablehttp", + "MCPGATEWAY_SKIP_MIGRATIONS": "true", + "GATEWAY_TOOL_NAME_SEPARATOR": "_", + "SSRF_ALLOW_LOCALHOST": "true", + "SSRF_ALLOW_PRIVATE_NETWORKS": "true", + "SSRF_DNS_FAIL_CLOSED": "false", + "PLUGINS_ENABLED": "false", + "MCPGATEWAY_CATALOG_ENABLED": "false", + "MCPGATEWAY_UI_ENABLED": "false", + "MCPGATEWAY_ADMIN_API_ENABLED": "true", + "ENABLE_METRICS": "false", + "DB_METRICS_RECORDING_ENABLED": "false", + "STRUCTURED_LOGGING_DATABASE_ENABLED": "false", + "AUDIT_TRAIL_ENABLED": "false", + "SECURITY_LOGGING_ENABLED": "false", + "DISABLE_ACCESS_LOG": "true", + "COMPRESSION_ENABLED": "false", + "VALIDATION_MIDDLEWARE_ENABLED": "false", + "CORRELATION_ID_ENABLED": "false", + "OBSERVABILITY_ENABLED": "false", + "RATE_LIMITING_ENABLED": "false", + "GUNICORN_WORKERS": "4", + "GUNICORN_KEEP_ALIVE": "30", + "GUNICORN_BACKLOG": "4096", + "DB_POOL_CLASS": "queue", + "DB_POOL_SIZE": "15", + "DB_MAX_OVERFLOW": "10", + "DB_POOL_PRE_PING": "true", + "HTTPX_MAX_CONNECTIONS": "1000", + "HTTPX_MAX_KEEPALIVE_CONNECTIONS": "500", + "MCP_SESSION_POOL_ENABLED": "true", + "MCP_SESSION_POOL_MAX_PER_KEY": "1000", + "TOOL_RATE_LIMIT": "600000", + "TOOL_CONCURRENT_LIMIT": "5000", + } + + +def deploy_builtin( + oc: Oc, + config: dict, + namespace: str, + node: str, + users: int | None = None, +) -> tuple[str, list[str]]: + database_name = with_users("builtin-db", users) + target_name = instance_name("target", "builtin", users) + migration_name = with_users("builtin-migration", users) + registration_name = with_users("builtin-registration", users) + total_cpu, total_memory = target_size(config) + request_cpu, request_memory = target_request_size(config) + postgres_cpu = total_cpu * 3 // 16 + redis_cpu = total_cpu // 16 + gateway_cpu = total_cpu - postgres_cpu - redis_cpu + postgres_memory = total_memory * 3 // 16 + redis_memory = total_memory // 16 + gateway_memory = total_memory - postgres_memory - redis_memory + postgres_request_cpu = request_cpu * 3 // 16 + redis_request_cpu = request_cpu // 16 + gateway_request_cpu = request_cpu - postgres_request_cpu - redis_request_cpu + postgres_request_memory = request_memory * 3 // 16 + redis_request_memory = request_memory // 16 + gateway_request_memory = ( + request_memory - postgres_request_memory - redis_request_memory + ) + credentials = { + "postgres": secrets.token_hex(24), + "jwt": secrets.token_hex(32), + "encryption": secrets.token_hex(32), + "password": secrets.token_hex(24), + } + environment = builtin_environment(credentials, database_name) + oc.apply( + { + "apiVersion": "v1", + "kind": "Pod", + "metadata": { + **metadata(database_name, namespace), + "labels": {"cf.contextforge/service": database_name}, + }, + "spec": { + "serviceAccountName": "benchmark", + "nodeName": node, + "restartPolicy": "Always", + "containers": [ + { + "name": "postgres", + "image": config["images"]["postgres"], + "args": [ + "postgres", + "-c", + "max_connections=300", + "-c", + "shared_buffers=256MB", + "-c", + "synchronous_commit=off", + ], + "env": env_list( + { + "POSTGRES_USER": "postgres", + "POSTGRES_PASSWORD": credentials["postgres"], + "POSTGRES_DB": "mcp", + } + ), + "ports": [{"containerPort": 5432}], + "resources": resources( + f"{postgres_cpu}m", + f"{postgres_memory}Mi", + f"{postgres_request_cpu}m", + f"{postgres_request_memory}Mi", + ), + "readinessProbe": { + "exec": { + "command": ["pg_isready", "-U", "postgres", "-d", "mcp"] + }, + "periodSeconds": 2, + }, + }, + { + "name": "redis", + "image": config["images"]["redis"], + "args": [ + "redis-server", + "--save", + "", + "--appendonly", + "no", + "--maxmemory", + "512mb", + "--maxmemory-policy", + "allkeys-lru", + ], + "ports": [{"containerPort": 6379}], + "resources": resources( + f"{redis_cpu}m", + f"{redis_memory}Mi", + f"{redis_request_cpu}m", + f"{redis_request_memory}Mi", + ), + "readinessProbe": { + "exec": {"command": ["redis-cli", "ping"]}, + "periodSeconds": 2, + }, + }, + ], + }, + } + ) + service( + oc, + namespace, + database_name, + database_name, + [ + {"name": "postgres", "port": 5432, "targetPort": 5432}, + {"name": "redis", "port": 6379, "targetPort": 6379}, + ], + ) + wait_for(oc, namespace, "pod", database_name, "condition=Ready", 300) + migration_env = {**environment, "MCPGATEWAY_SKIP_MIGRATIONS": "false"} + oc.apply( + { + "apiVersion": "v1", + "kind": "Pod", + "metadata": metadata(migration_name, namespace), + "spec": { + "serviceAccountName": "benchmark", + "nodeName": node, + "restartPolicy": "Never", + "containers": [ + { + "name": "migration", + "image": config["images"]["controlplane"], + "command": ["python3", "-m", "mcpgateway.bootstrap_db"], + "env": env_list(migration_env), + "resources": resources("500m", "512Mi"), + } + ], + }, + } + ) + wait_for(oc, namespace, "pod", migration_name, "phase=Succeeded", 600) + oc.delete("pod", migration_name, "-n", namespace) + oc.apply( + { + "apiVersion": "v1", + "kind": "Pod", + "metadata": { + **metadata(target_name, namespace), + "labels": {"cf.contextforge/service": target_name}, + }, + "spec": { + "serviceAccountName": "benchmark", + "nodeName": node, + "restartPolicy": "Always", + "containers": [ + { + "name": "gateway", + "image": config["images"]["controlplane"], + "env": env_list({**environment, "HOST": "0.0.0.0", "PORT": "4444"}), + "ports": [{"containerPort": 4444}], + "resources": resources( + f"{gateway_cpu}m", + f"{gateway_memory}Mi", + f"{gateway_request_cpu}m", + f"{gateway_request_memory}Mi", + ), + "readinessProbe": { + "httpGet": {"path": "/health", "port": 4444}, + "periodSeconds": 3, + "failureThreshold": 100, + }, + } + ], + }, + } + ) + service( + oc, + namespace, + target_name, + target_name, + [{"name": "mcp", "port": 4444, "targetPort": 4444}], + ) + wait_for(oc, namespace, "pod", target_name, "condition=Ready", 600) + oc.apply( + { + "apiVersion": "v1", + "kind": "Pod", + "metadata": metadata(registration_name, namespace), + "spec": { + "serviceAccountName": "benchmark", + "nodeName": node, + "restartPolicy": "Never", + "volumes": [{"name": "code", "configMap": {"name": "benchmark-code"}}], + "containers": [ + { + "name": "registration", + "image": config["images"]["controlplane"], + "command": ["python3", "/work/register_builtin.py"], + "args": [ + "--backend", + f"http://{instance_name('fast-time', 'builtin', users)}:9080/mcp", + ], + "env": env_list( + { + **environment, + "GATEWAY_URL": f"http://{target_name}:4444", + } + ), + "volumeMounts": [{"name": "code", "mountPath": "/work"}], + "resources": resources("250m", "256Mi"), + } + ], + }, + } + ) + wait_for(oc, namespace, "pod", registration_name, "phase=Succeeded", 600) + output = oc.run( + "logs", registration_name, "-n", namespace, "-c", "registration" + ).stdout.splitlines() + if not output: + raise RuntimeError("built-in registration produced no result") + registration = json.loads(output[-1]) + oc.delete("pod", registration_name, "-n", namespace) + return registration["token"], registration["tool_names"] + + +def store_lane_secret( + oc: Oc, + namespace: str, + lane: str, + token: str, + tools: list[str], + users: int, +) -> None: + oc.apply( + { + "apiVersion": "v1", + "kind": "Secret", + "metadata": metadata(instance_name("lane", lane, users), namespace), + "type": "Opaque", + "stringData": {"token": token, "tools": ",".join(tools)}, + } + ) + + +def smoke_lane( + oc: Oc, + config: dict, + namespace: str, + lane: str, + node: str, + url: str, + tools: list[str], + users: int, +) -> None: + name = instance_name("smoke", lane, users) + oc.delete("pod", name, "-n", namespace) + oc.apply( + { + "apiVersion": "v1", + "kind": "Pod", + "metadata": metadata(name, namespace), + "spec": { + "serviceAccountName": "benchmark", + "nodeName": node, + "restartPolicy": "Never", + "volumes": [ + {"name": "code", "configMap": {"name": "benchmark-code"}}, + { + "name": "secret", + "secret": { + "secretName": instance_name("lane", lane, users) + }, + }, + ], + "containers": [ + { + "name": "smoke", + "image": config["images"]["locust"], + "command": ["python"], + "args": [ + "/work/smoke.py", + "--urls", + url, + "--token-file", + "/secret/token", + "--tool-names", + ",".join(tools), + ], + "volumeMounts": [ + {"name": "code", "mountPath": "/work"}, + {"name": "secret", "mountPath": "/secret", "readOnly": True}, + ], + "resources": resources("500m", "512Mi"), + } + ], + }, + } + ) + wait_for(oc, namespace, "pod", name, "phase=Succeeded", 300) + oc.delete("pod", name, "-n", namespace) + + +def load_environment( + config: dict, lane: str, url: str, tools: list[str], users: int +) -> list[dict]: + values = { + "MCP_PROTOCOL_VERSION": config["workload"]["protocol_version"], + "MCP_STACK_MODE": "controlplane" if lane == "builtin" else "dataplane", + "MCP_SERVER_ID": SERVER_ID, + "MCP_DIRECT_DATAPLANE": "false" if lane == "builtin" else "true", + "MCP_SKIP_TOOL_LIST": "true", + "MCP_EXPLICIT_ZERO_DELAY": "true", + "MCP_FYRE_WORKLOAD": "true", + "MCP_TOOL_NAMES": ",".join(tools), + "MCP_BASE_URLS": url, + "LOCUST_REQUEST_TIMEOUT_SECONDS": "30", + "MCP_MEASUREMENT_MARKER": "/reports/measurement-start.txt", + "MCP_MEASUREMENT_SECONDS": str(config["workload"]["measure_seconds"]), + "MCP_WARMUP_SECONDS": str(config["workload"]["warmup_seconds"]), + } + return [*env_list(values), secret_env(instance_name("lane", lane, users))] + + +def locust_pod( + config: dict, + namespace: str, + lane: str, + node: str, + users: int, + url: str, + tools: list[str], +) -> dict: + name = f"load-{lane}-{users}" + workload = config["workload"] + total = ( + int(workload["ramp_seconds"]) + + int(workload["warmup_seconds"]) + + int(workload["measure_seconds"]) + + 30 + ) + common_mounts = [ + {"name": "code", "mountPath": "/mnt/locust-cf"}, + {"name": "reports", "mountPath": "/reports"}, + ] + common_env = load_environment(config, lane, url, tools, users) + master_args = [ + "-f", + "/mnt/locust-cf/locustfile_mcp.py", + "--master", + "--expect-workers", + "3", + "--headless", + "--users", + str(users), + "--spawn-rate", + str(max(1.0, users / int(workload["ramp_seconds"]))), + "--run-time", + f"{total}s", + "--stop-timeout", + "1", + "--host", + "http://127.0.0.1", + "--csv", + "/reports/locust", + "--csv-full-history", + "--html", + "/reports/locust.html", + "--json-file", + "/reports/locust.json", + "--logfile", + "/reports/locust.log", + ] + containers = [ + { + "name": "master", + "image": config["images"]["locust"], + "args": master_args, + "env": common_env, + "volumeMounts": common_mounts, + "resources": resources( + f"{int(config['infrastructure']['openshift']['load_pod']['cpu_millicores']) // 4}m", + f"{int(config['infrastructure']['openshift']['load_pod']['memory_mib']) // 4}Mi", + ), + } + ] + for index in range(3): + containers.append( + { + "name": f"worker-{index + 1}", + "image": config["images"]["locust"], + "args": [ + "-f", + "/mnt/locust-cf/locustfile_mcp.py", + "--worker", + "--master-host", + "127.0.0.1", + ], + "env": [ + *common_env, + {"name": "MCP_REPLICA_OFFSET", "value": str(index)}, + ], + "volumeMounts": common_mounts, + "resources": resources( + f"{int(config['infrastructure']['openshift']['load_pod']['cpu_millicores']) // 4}m", + f"{int(config['infrastructure']['openshift']['load_pod']['memory_mib']) // 4}Mi", + ), + } + ) + return { + "apiVersion": "v1", + "kind": "Pod", + "metadata": metadata(name, namespace), + "spec": { + "serviceAccountName": "benchmark", + "nodeName": node, + "restartPolicy": "Never", + "terminationGracePeriodSeconds": 5, + "volumes": [ + {"name": "code", "configMap": {"name": "benchmark-code"}}, + {"name": "reports", "emptyDir": {}}, + ], + "containers": containers, + }, + } + + +def terminated_status(pod: dict, container: str) -> tuple[bool, int]: + for status in pod.get("status", {}).get("containerStatuses", []): + if status.get("name") != container: + continue + terminated = status.get("state", {}).get("terminated") + if terminated: + return True, int(terminated.get("exitCode", 1)) + return False, 0 + + +def telemetry_snapshot(oc: Oc, namespace: str) -> dict: + timestamp = time.time() + pods = oc.run( + "adm", "top", "pods", "-n", namespace, "--containers", "--no-headers", check=False + ) + nodes = oc.run("adm", "top", "nodes", "--no-headers", check=False) + return { + "time": timestamp, + "pods": pods.stdout, + "pods_error": pods.stderr if pods.returncode else "", + "nodes": nodes.stdout, + "nodes_error": nodes.stderr if nodes.returncode else "", + } + + +def parse_memory_mib(value: str) -> float: + units = {"Ki": 1 / 1024, "Mi": 1.0, "Gi": 1024.0} + for suffix, multiplier in units.items(): + if value.endswith(suffix): + return float(value[: -len(suffix)]) * multiplier + return float(value) / 1024 / 1024 + + +def parse_cpu_millicores(value: str) -> float: + units = {"n": 1 / 1_000_000, "u": 1 / 1_000, "m": 1.0} + for suffix, multiplier in units.items(): + if value.endswith(suffix): + return float(value[: -len(suffix)]) * multiplier + return float(value) * 1000 + + +def measurement_samples( + samples: list[dict], start_time: float | None, end_time: float | None +) -> list[dict]: + selected = [] + for sample in samples: + timestamp = float(sample.get("time", 0)) + if start_time is not None and timestamp < start_time: + continue + if end_time is not None and timestamp > end_time: + continue + selected.append(sample) + return selected + + +def lane_memory( + samples: list[dict], + lane: str, + users: int | None = None, + start_time: float | None = None, + end_time: float | None = None, +) -> dict[str, float]: + pod_names = ( + ( + instance_name("target", "builtin", users), + with_users("builtin-db", users), + ) + if lane == "builtin" + else (instance_name("target", "external", users),) + ) + totals = [] + for sample in measurement_samples(samples, start_time, end_time): + total = 0.0 + seen = False + for line in sample.get("pods", "").splitlines(): + fields = line.split() + if len(fields) < 4 or fields[0] not in pod_names: + continue + try: + total += parse_memory_mib(fields[3]) + seen = True + except ValueError: + continue + if seen: + totals.append(total) + return { + "average_mib": sum(totals) / len(totals) if totals else 0.0, + "peak_mib": max(totals, default=0.0), + "samples": len(totals), + } + + +def helper_pressure( + samples: list[dict], + lane: str, + config: dict, + users: int, + start_time: float, + end_time: float, +) -> dict: + load_size = config["infrastructure"]["openshift"]["load_pod"] + backend_size = config["infrastructure"]["openshift"]["backend_pod"] + load_cpu = float(load_size["cpu_millicores"]) + load_memory = float(load_size["memory_mib"]) + backend_cpu = float(backend_size["cpu_millicores"]) + backend_memory = float(backend_size["memory_mib"]) + snapshots = measurement_samples(samples, start_time, end_time) + totals: list[tuple[float, float, float, float]] = [] + worker_cpu: dict[str, list[float]] = {} + for sample in snapshots: + locust_cpu = locust_memory = fast_cpu = fast_memory = 0.0 + seen_locust = seen_fast = False + for line in sample.get("pods", "").splitlines(): + fields = line.split() + if len(fields) < 4: + continue + pod, container, cpu, memory = fields[:4] + try: + cpu_milli = parse_cpu_millicores(cpu) + memory_mib = parse_memory_mib(memory) + except ValueError: + continue + if pod == instance_name("load", lane, users): + locust_cpu += cpu_milli + locust_memory += memory_mib + seen_locust = True + if container.startswith("worker-"): + worker_cpu.setdefault(container, []).append( + cpu_milli / (load_cpu / 4) * 100 + ) + elif pod == instance_name("fast-time", lane, users): + fast_cpu += cpu_milli + fast_memory += memory_mib + seen_fast = True + if seen_locust and seen_fast: + totals.append( + ( + locust_cpu / load_cpu * 100, + locust_memory / load_memory * 100, + fast_cpu / backend_cpu * 100, + fast_memory / backend_memory * 100, + ) + ) + average = lambda index: ( + sum(values[index] for values in totals) / len(totals) if totals else 0.0 + ) + worker_max = max( + (sum(values) / len(values) for values in worker_cpu.values()), default=0.0 + ) + thresholds = config["workload"] + saturated = [] + if len(totals) < 3: + saturated.append("helper metrics unavailable") + if average(0) > float(thresholds["helper_cpu_percent"]): + saturated.append("Locust CPU") + if average(1) > float(thresholds["helper_memory_percent"]): + saturated.append("Locust memory") + if average(2) > float(thresholds["helper_cpu_percent"]): + saturated.append("Fast Time CPU") + if average(3) > float(thresholds["helper_memory_percent"]): + saturated.append("Fast Time memory") + if worker_max > float(thresholds["worker_core_percent"]): + saturated.append("Locust worker CPU") + return { + "samples": len(totals), + "locust_cpu_average_percent": average(0), + "locust_memory_average_percent": average(1), + "fast_time_cpu_average_percent": average(2), + "fast_time_memory_average_percent": average(3), + "worker_cpu_max_average_percent": worker_max, + "saturated": saturated, + } + + +def collect_reports( + oc: Oc, namespace: str, pod: str, destination: Path +) -> None: + destination.mkdir(parents=True, exist_ok=True) + result = oc.run( + "cp", + f"{namespace}/{pod}:/reports/.", + oc.container_path(destination), + "-c", + "master", + check=False, + timeout=300, + ) + if result.returncode: + raise RuntimeError(f"failed to collect {pod} reports: {result.stderr}") + + +def run_parallel_step( + oc: Oc, + config: dict, + namespace: str, + nodes: dict[str, list[str]], + urls: dict[str, str], + tools: dict[str, list[str]], + output: Path, + users: int, + stop_event: threading.Event | None = None, +) -> dict[str, dict]: + pods = {lane: f"load-{lane}-{users}" for lane in LANES} + for lane in LANES: + oc.delete("pod", pods[lane], "-n", namespace) + oc.apply( + locust_pod( + config, + namespace, + lane, + assigned_node(nodes, "locust", lane, users), + users, + urls[lane], + tools[lane], + ) + ) + completed: dict[str, int] = {} + samples: list[dict] = [] + deadline = time.monotonic() + ( + int(config["workload"]["ramp_seconds"]) + + int(config["workload"]["warmup_seconds"]) + + int(config["workload"]["measure_seconds"]) + + 300 + ) + last_sample = 0.0 + while len(completed) < len(LANES): + if stop_event is not None and stop_event.is_set(): + break + if time.monotonic() > deadline: + raise RuntimeError(f"parallel {users}-user step exceeded its time bound") + for lane, pod_name in pods.items(): + if lane in completed: + continue + pod = oc.json("get", "pod", pod_name, "-n", namespace) + for container in ["master", "worker-1", "worker-2", "worker-3"]: + done, exit_code = terminated_status(pod, container) + if done and exit_code and container != "master": + completed[lane] = exit_code + break + done, exit_code = terminated_status(pod, "master") + if done: + completed[lane] = exit_code + now = time.monotonic() + if now - last_sample >= 10: + samples.append(telemetry_snapshot(oc, namespace)) + last_sample = now + if any(code for code in completed.values()): + if stop_event is not None: + stop_event.set() + break + time.sleep(2) + + results: dict[str, dict] = {} + for lane, pod_name in pods.items(): + lane_output = output / f"{lane}-{users}" + try: + collect_reports(oc, namespace, pod_name, lane_output) + marker = lane_output / "measurement-start.txt" + if not marker.is_file(): + raise RuntimeError("Locust did not record the measurement-window start") + measurement_start = float(marker.read_text(encoding="utf-8").strip()) + measurement_end = measurement_start + int( + config["workload"]["measure_seconds"] + ) + stats = read_stats(lane_output / "locust_stats.csv", use_aggregate=True) + exit_code = completed.get(lane, 1) + pressure = helper_pressure( + samples, lane, config, users, measurement_start, measurement_end + ) + stats.update( + { + "lane": "rust" if lane == "external" else "builtin", + "users": users, + "passed": exit_code == 0 + and stats["failures"] == 0 + and not pressure["saturated"], + "memory": lane_memory( + samples, lane, users, measurement_start, measurement_end + ), + "pressure": pressure, + } + ) + if exit_code: + stats["reason"] = f"Locust exited {exit_code}" + elif pressure["saturated"]: + stats["reason"] = "helper pressure: " + ", ".join( + pressure["saturated"] + ) + if stop_event is not None: + stop_event.set() + results[lane] = stats + except Exception as error: + results[lane] = { + "lane": "rust" if lane == "external" else "builtin", + "users": users, + "passed": False, + "reason": str(error), + "memory": lane_memory(samples, lane, users), + "pressure": {}, + } + finally: + oc.delete("pod", pod_name, "-n", namespace) + telemetry = output / f"telemetry-{users}.jsonl" + telemetry.write_text( + "".join(json.dumps(sample, sort_keys=True) + "\n" for sample in samples), + encoding="utf-8", + ) + return results + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("--config", required=True) + parser.add_argument("--kubeconfig", required=True) + parser.add_argument("--output", required=True) + parser.add_argument("--run-id", required=True) + parser.add_argument("--assets", required=True) + args = parser.parse_args() + config = json.loads(Path(args.config).read_text(encoding="utf-8")) + output = Path(args.output) + output.mkdir(parents=True, exist_ok=True) + openshift = config["infrastructure"]["openshift"] + oc = Oc(openshift["oc_image"], Path(args.kubeconfig)) + namespace = f"cf-fyre-{args.run_id}" + result = { + "status": "running", + "scenario": config["scenarios"][0], + "protocol_version": config["workload"]["protocol_version"], + "user_levels": config["workload"]["user_levels"], + "parallel_lanes": True, + "runs": {"rust": [], "builtin": []}, + } + result_path = output / "result.json" + + def save() -> None: + result_path.write_text( + json.dumps(result, indent=2, sort_keys=True) + "\n", encoding="utf-8" + ) + + save() + try: + setup_namespace(oc, namespace, Path(args.assets)) + nodes = assign_nodes(config, oc.json("get", "nodes")) + all_parallel = bool(config["workload"].get("parallel_user_levels")) + result["inventory"] = { + "namespace": namespace, + "nodes": nodes, + "architecture": ( + "eight reserved 2v2 targets running concurrently on three dedicated-role workers" + if all_parallel + else "two isolated lanes running concurrently on six workers" + ), + } + (output / "inventory.json").write_text( + json.dumps(result["inventory"], indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + levels = [int(users) for users in config["workload"]["user_levels"]] + instances = ( + [(lane, users) for users in levels for lane in LANES] + if all_parallel + else [(lane, None) for lane in LANES] + ) + with concurrent.futures.ThreadPoolExecutor( + max_workers=len(instances) + ) as executor: + futures = [ + executor.submit( + deploy_fast_time, + oc, + config, + namespace, + lane, + assigned_node(nodes, "fast-time", lane, users), + users, + ) + for lane, users in instances + ] + for future in futures: + future.result() + credentials: dict[tuple[str, int | None], tuple[str, list[str]]] = {} + with concurrent.futures.ThreadPoolExecutor( + max_workers=len(instances) + ) as executor: + futures = {} + for lane, users in instances: + function = deploy_builtin if lane == "builtin" else deploy_external + future = executor.submit( + function, + oc, + config, + namespace, + assigned_node(nodes, "target", lane, users), + users, + ) + futures[future] = (lane, users) + for future, key in futures.items(): + credentials[key] = future.result() + urls: dict[tuple[str, int], str] = {} + for users in levels: + for lane in LANES: + credential_key = (lane, users if all_parallel else None) + token, tool_names = credentials[credential_key] + if len(tool_names) != 6: + raise RuntimeError( + f"{lane} at {users} users did not expose all six benchmark tools" + ) + store_lane_secret( + oc, namespace, lane, token, tool_names, users + ) + target = instance_name( + "target", lane, users if all_parallel else None + ) + urls[(lane, users)] = ( + f"http://{target}:4444/mcp" + if lane == "builtin" + else f"http://{target}:4445/contextforge-rs/servers/{SERVER_ID}/mcp" + ) + smoke_instances = ( + [(lane, users) for users in levels for lane in LANES] + if all_parallel + else [(lane, levels[0]) for lane in LANES] + ) + with concurrent.futures.ThreadPoolExecutor( + max_workers=len(smoke_instances) + ) as executor: + futures = [ + executor.submit( + smoke_lane, + oc, + config, + namespace, + lane, + assigned_node(nodes, "locust", lane, users), + urls[(lane, users)], + credentials[(lane, users if all_parallel else None)][1], + users, + ) + for lane, users in smoke_instances + ] + for future in futures: + future.result() + stop_event = threading.Event() + step_results: dict[int, dict[str, dict]] = {} + workers = len(levels) if all_parallel else 1 + with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as executor: + futures = {} + for users in levels: + future = executor.submit( + run_parallel_step, + oc, + config, + namespace, + nodes, + {lane: urls[(lane, users)] for lane in LANES}, + { + lane: credentials[ + (lane, users if all_parallel else None) + ][1] + for lane in LANES + }, + output, + users, + stop_event if all_parallel else None, + ) + futures[future] = users + if not all_parallel: + step = future.result() + step_results[users] = step + if not all(item.get("passed") for item in step.values()): + break + if all_parallel: + for future, users in futures.items(): + step_results[users] = future.result() + for users in sorted(step_results): + step = step_results[users] + result["runs"]["builtin"].append(step["builtin"]) + result["runs"]["rust"].append(step["external"]) + save() + if len(step_results) != len(levels) or not all( + item.get("passed") + for step in step_results.values() + for item in step.values() + ): + result["status"] = "failed" + result["reason"] = "first request, worker, or helper-pressure error" + save() + raise SystemExit(1) + result["status"] = "confirmed" + save() + except BaseException as error: + if result["status"] == "running": + result["status"] = "failed" + result["reason"] = str(error) + save() + raise + finally: + oc.delete("namespace", namespace, "--wait=false") + + +if __name__ == "__main__": + main() diff --git a/benchmarks/fyre/report.py b/benchmarks/fyre/report.py index b8f37c7..8d93785 100644 --- a/benchmarks/fyre/report.py +++ b/benchmarks/fyre/report.py @@ -8,6 +8,147 @@ from pathlib import Path +def comparison_markdown(config: dict, rows: list[dict]) -> str: + infrastructure = config.get("infrastructure", {}) + openshift = infrastructure.get("kind") == "openshift" + openshift_config = infrastructure.get("openshift", {}) + workload = config.get("workload", {}) + all_parallel = openshift and bool(workload.get("parallel_user_levels")) + target = (config.get("scenarios") or [{}])[0] + images = config.get("images", {}) + tools = workload.get("tools", []) + duration = int(workload.get("measure_seconds", 0)) + helpers = config.get("active_helper", {}) + locust_cpu = ( + openshift_config.get("load_pod", {}).get("cpu_millicores", 0) / 1_000 + if openshift + else helpers.get("locust_cpu", "n/a") + ) + locust_memory = ( + openshift_config.get("load_pod", {}).get("memory_mib", 0) / 1_024 + if openshift + else helpers.get("locust_memory_gb", "n/a") + ) + fast_time_cpu = ( + openshift_config.get("backend_pod", {}).get("cpu_millicores", 0) / 1_000 + if openshift + else helpers.get("fast_time_cpu", "n/a") + ) + fast_time_memory = ( + openshift_config.get("backend_pod", {}).get("memory_mib", 0) / 1_024 + if openshift + else helpers.get("fast_time_memory_gb", "n/a") + ) + disk = ( + f"{infrastructure.get('openshift', {}).get('base_disk_gb', 'n/a')} GB" + if openshift + else "250 GB (standalone FYRE default)" + ) + lines = [ + "# FYRE built-in dataplane and external dataplane comparison", + "", + "## Architecture", + "", + "```mermaid", + "flowchart LR", + " subgraph BI[Built-in dataplane lane]", + f" LBI[Locust: {locust_cpu} vCPU / {locust_memory} GiB] --> TBI[Gateway + PostgreSQL + Redis: {target.get('cpu', 'n/a')} vCPU / {target.get('memory_gb', 'n/a')} GiB]", + f" TBI --> FBI[Fast Time: {fast_time_cpu} vCPU / {fast_time_memory} GiB]", + " end", + " subgraph EX[External dataplane lane]", + f" LEX[Locust: {locust_cpu} vCPU / {locust_memory} GiB] --> TEX[Rust + Redis + JWKS: {target.get('cpu', 'n/a')} vCPU / {target.get('memory_gb', 'n/a')} GiB]", + f" TEX --> FEX[Fast Time: {fast_time_cpu} vCPU / {fast_time_memory} GiB]", + " end", + "```", + "", + ( + ( + "All eight lane/user measurements ran concurrently. Each measurement had " + "its own reserved Locust, target, and Fast Time pods. Pods of the same role " + "shared a dedicated worker pool, with requests and limits equal to the stated " + "per-measurement allocation." + if all_parallel + else "The two lanes ran concurrently on six dedicated OpenShift worker nodes. " + "Each lane had its own load generator, target, and backend, so the measured " + "targets and helpers shared no worker node." + ) + if openshift + else "The two lanes ran sequentially on the same standalone target VM." + ), + "", + "| Setting | Value |", + "| --- | --- |", + f"| Infrastructure | {'FYRE OpenShift ' + str(infrastructure.get('openshift', {}).get('version', '')) if openshift else 'FYRE standalone VMs'} |", + f"| Worker root disk | {disk} |", + f"| Target allocation per lane | {target.get('cpu', 'n/a')} vCPU / {target.get('memory_gb', 'n/a')} GiB |", + f"| Locust allocation per measurement | {locust_cpu} vCPU / {locust_memory} GiB; " + + ("one master and three workers |" if openshift else "distributed workers |"), + f"| Fast Time allocation per measurement | {fast_time_cpu} vCPU / {fast_time_memory} GiB |", + f"| MCP protocol | {workload.get('protocol_version', 'n/a')} |", + f"| Timing per measurement | {workload.get('ramp_seconds', 0)} s ramp, {workload.get('warmup_seconds', 0)} s warmup, {duration // 60} min measured |", + "| Client | Locust FastHttpUser, zero wait |", + "| Failure policy | Stop all concurrent measurements on the first request or worker error |" + if all_parallel + else "| Failure policy | Stop the current pair on the first request or worker error; do not advance |", + "", + "## Requests", + "", + "Each virtual user calls `server/discover` once during startup. Statistics reset after warmup, so measured traffic consists only of repeated `tools/call` requests with uniform random selection across these six Fast Time tools:", + "", + ", ".join(f"`{tool}`" for tool in tools) + ".", + "", + "The backend has explicit zero delay. The workload makes no `resources/read` requests and no fan-out calls.", + "", + "## Results", + "", + "| Users | Built-in requests | Built-in errors | Built-in RPS | Built-in p50/p95/p99 | Built-in memory avg/peak | External requests | External errors | External RPS | External p50/p95/p99 | External memory avg/peak | External vs built-in |", + "| ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |", + ] + for row in rows: + builtin_memory = memory_pair( + row["built_in_dataplane_memory_average_mib"], + row["built_in_dataplane_memory_peak_mib"], + ) + external_memory = memory_pair( + row["external_dataplane_memory_average_mib"], + row["external_dataplane_memory_peak_mib"], + ) + lines.append( + f"| {row['users']:,} | {row['built_in_dataplane_requests']:,} | " + f"{row['built_in_dataplane_errors']} | {row['built_in_dataplane_rps']:,.2f} | " + f"{row['built_in_dataplane_p50_ms']:.0f}/{row['built_in_dataplane_p95_ms']:.0f}/{row['built_in_dataplane_p99_ms']:.0f} ms | " + f"{builtin_memory} | " + f"{row['external_dataplane_requests']:,} | {row['external_dataplane_errors']} | " + f"{row['external_dataplane_rps']:,.2f} | " + f"{row['external_dataplane_p50_ms']:.0f}/{row['external_dataplane_p95_ms']:.0f}/{row['external_dataplane_p99_ms']:.0f} ms | " + f"{external_memory} | " + f"{row['external_vs_built_in']:.2f}× |" + ) + lines.extend( + [ + "", + "Gateway memory is sampled every 10 seconds during the measured window and includes all gateway-side containers: Python, PostgreSQL, and Redis for the built-in lane; Rust, Redis, and loopback JWKS for the external lane.", + "", + "## Software", + "", + "| Component | Pinned image |", + "| --- | --- |", + *[ + f"| {name.replace('_', ' ').title()} | `{image}` |" + for name, image in sorted(images.items()) + ], + "", + ] + ) + return "\n".join(lines) + + +def memory_pair(average: float | None, peak: float | None) -> str: + if average is None or peak is None: + return "n/a" + return f"{average:.0f}/{peak:.0f} MiB" + + def comparison_report(config: dict, results_root: Path, *, render: bool = True) -> None: result_path = results_root / "comparison" / "result.json" if not result_path.is_file(): @@ -32,12 +173,16 @@ def comparison_report(config: dict, results_root: Path, *, render: bool = True) "built_in_dataplane_p50_ms": builtin["p50_ms"], "built_in_dataplane_p95_ms": builtin["p95_ms"], "built_in_dataplane_p99_ms": builtin["p99_ms"], + "built_in_dataplane_memory_average_mib": builtin.get("memory", {}).get("average_mib"), + "built_in_dataplane_memory_peak_mib": builtin.get("memory", {}).get("peak_mib"), "external_dataplane_requests": rust["requests"], "external_dataplane_errors": rust["failures"], "external_dataplane_rps": rust["rps"], "external_dataplane_p50_ms": rust["p50_ms"], "external_dataplane_p95_ms": rust["p95_ms"], "external_dataplane_p99_ms": rust["p99_ms"], + "external_dataplane_memory_average_mib": rust.get("memory", {}).get("average_mib"), + "external_dataplane_memory_peak_mib": rust.get("memory", {}).get("peak_mib"), "external_vs_built_in": rust["rps"] / builtin["rps"], } ) @@ -48,6 +193,9 @@ def comparison_report(config: dict, results_root: Path, *, render: bool = True) writer = csv.DictWriter(stream, fieldnames=rows[0].keys()) writer.writeheader() writer.writerows(rows) + (results_root / "report.md").write_text( + comparison_markdown(config, rows), encoding="utf-8" + ) if not render: return @@ -55,16 +203,27 @@ def comparison_report(config: dict, results_root: Path, *, render: bool = True) import matplotlib.pyplot as plt from matplotlib.patches import FancyBboxPatch - helpers = config["active_helper"] + openshift = config.get("infrastructure", {}).get("kind") == "openshift" + openshift_config = config.get("infrastructure", {}).get("openshift", {}) + helpers = config.get("active_helper", {}) workload = config["workload"] + all_parallel = openshift and bool(workload.get("parallel_user_levels")) target = config["scenarios"][0] + load_cpu = openshift_config.get("load_pod", {}).get("cpu_millicores", 0) / 1_000 + load_memory = openshift_config.get("load_pod", {}).get("memory_mib", 0) / 1_024 + backend_cpu = openshift_config.get("backend_pod", {}).get("cpu_millicores", 0) / 1_000 + backend_memory = openshift_config.get("backend_pod", {}).get("memory_mib", 0) / 1_024 figure = plt.figure(figsize=(18, 10), dpi=160, facecolor="#0b1020") axis = figure.add_axes([0, 0, 1, 1]) axis.set_axis_off() figure.text( 0.035, 0.95, - "FYRE built-in dataplane vs external dataplane — one-hour load comparison", + "FYRE OpenShift built-in vs external dataplane — eight parallel one-hour measurements" + if all_parallel + else "FYRE OpenShift built-in vs external dataplane — parallel one-hour load comparison" + if openshift + else "FYRE built-in dataplane vs external dataplane — one-hour load comparison", color="white", fontsize=25, fontweight="bold", @@ -72,7 +231,11 @@ def comparison_report(config: dict, results_root: Path, *, render: bool = True) figure.text( 0.035, 0.91, - "Eight zero-error benchmarks • same modern client and same target VM allocation • private FYRE network", + "Eight benchmarks run together • reserved 2 vCPU / 2 GB targets • 40 GB OpenShift nodes" + if all_parallel + else "Eight zero-error benchmarks • isolated equal-size lane allocations • 40 GB OpenShift nodes" + if openshift + else "Eight zero-error benchmarks • same modern client and same target VM allocation • private FYRE network", color="#a7b0c0", fontsize=12, ) @@ -81,20 +244,38 @@ def comparison_report(config: dict, results_root: Path, *, render: bool = True) ( 0.035, "LOAD GENERATOR", - f"Locust VM • {helpers['locust_cpu']} vCPU / {helpers['locust_memory_gb']} GB\n" + f"Per measurement: {load_cpu:g} vCPU / {load_memory:g} GB pod\n" + "3 distributed workers • zero wait\nShared dedicated-role OpenShift worker" + if all_parallel + else "Per lane: 4 vCPU / 16 GB pod\n3 distributed workers • zero wait\nDedicated OpenShift worker" + if openshift + else f"Locust VM • {helpers['locust_cpu']} vCPU / {helpers['locust_memory_gb']} GB\n" f"{max(2, int(helpers['locust_cpu']) - 1)} distributed workers • zero wait", ), ( 0.355, - "TARGET — SAME VM, SEQUENTIAL", - f"{target['cpu']} vCPU / {target['memory_gb']} GB\n" + "TARGETS — 8 RESERVED PODS" + if all_parallel + else "TARGETS — ISOLATED, PARALLEL" + if openshift + else "TARGET — SAME VM, SEQUENTIAL", + f"Per measurement: {target['cpu']} vCPU / {target['memory_gb']} GB pod allocation\n" + "Built-in: Python + Postgres + Redis\nExternal: Rust + Redis + loopback JWKS\n" + "Shared dedicated target worker" + if openshift + else f"{target['cpu']} vCPU / {target['memory_gb']} GB\n" "Built-in dataplane: Python gateway + Postgres + Redis\n" "External dataplane: Rust + Redis + loopback JWKS", ), ( 0.71, "BACKEND", - f"Fast Time VM • {helpers['fast_time_cpu']} vCPU / {helpers['fast_time_memory_gb']} GB\n6 tools • explicit zero delay", + f"Per measurement: {backend_cpu:g} vCPU / {backend_memory:g} GB pod\n" + "6 tools • explicit zero delay\nShared dedicated-role OpenShift worker" + if all_parallel + else "Per lane: 8 vCPU / 32 GB pod\n6 tools • explicit zero delay\nDedicated OpenShift worker" + if openshift + else f"Fast Time VM • {helpers['fast_time_cpu']} vCPU / {helpers['fast_time_memory_gb']} GB\n6 tools • explicit zero delay", ), ] widths = [0.27, 0.31, 0.255] @@ -132,6 +313,8 @@ def comparison_report(config: dict, results_root: Path, *, render: bool = True) "External DP\nerrors", "External DP\nRPS", "External DP p50 /\np95 / p99", + "Built-in memory\navg / peak MiB", + "External memory\navg / peak MiB", "External vs\nbuilt-in", ] cells = [ @@ -145,6 +328,14 @@ def comparison_report(config: dict, results_root: Path, *, render: bool = True) str(row["external_dataplane_errors"]), f"{row['external_dataplane_rps']:,.2f}", f"{row['external_dataplane_p50_ms']:.0f} / {row['external_dataplane_p95_ms']:.0f} / {row['external_dataplane_p99_ms']:.0f} ms", + memory_pair( + row["built_in_dataplane_memory_average_mib"], + row["built_in_dataplane_memory_peak_mib"], + ).removesuffix(" MiB").replace("/", " / "), + memory_pair( + row["external_dataplane_memory_average_mib"], + row["external_dataplane_memory_peak_mib"], + ).removesuffix(" MiB").replace("/", " / "), f"{row['external_vs_built_in']:.2f}×", ] for row in rows @@ -154,10 +345,10 @@ def comparison_report(config: dict, results_root: Path, *, render: bool = True) colLabels=headers, cellLoc="center", loc="center", - colWidths=[0.06, 0.105, 0.065, 0.09, 0.15, 0.105, 0.065, 0.09, 0.15, 0.10], + colWidths=[0.05, 0.09, 0.055, 0.075, 0.115, 0.09, 0.055, 0.075, 0.115, 0.09, 0.09, 0.08], ) table.auto_set_font_size(False) - table.set_fontsize(9.5) + table.set_fontsize(8.0) table.scale(1, 2.3) for (row, _column), cell in table.get_celld().items(): cell.set_edgecolor("#34415f") diff --git a/benchmarks/fyre/test_campaign.py b/benchmarks/fyre/test_campaign.py index 92c88cf..ffb1c54 100644 --- a/benchmarks/fyre/test_campaign.py +++ b/benchmarks/fyre/test_campaign.py @@ -659,10 +659,13 @@ def test_comparison_report_writes_machine_readable_lane_results(self): {"workload": {"user_levels": [125]}}, root, render=False ) summary = json.loads((root / "summary.json").read_text()) + markdown = (root / "report.md").read_text() with (root / "summary.csv").open(newline="") as stream: csv_rows = list(csv.DictReader(stream)) self.assertEqual(summary["rows"][0]["external_vs_built_in"], 2.5) self.assertEqual(csv_rows[0]["external_dataplane_requests"], "2500") + self.assertIn("External vs built-in", markdown) + self.assertNotIn("](", markdown) def test_monitor_calculates_cpu_and_memory_pressure(self): cpu = monitor.cpu_percent( diff --git a/benchmarks/fyre/test_openshift_campaign.py b/benchmarks/fyre/test_openshift_campaign.py new file mode 100644 index 0000000..42000f0 --- /dev/null +++ b/benchmarks/fyre/test_openshift_campaign.py @@ -0,0 +1,202 @@ +"""Focused tests for the OpenShift benchmark topology.""" + +from __future__ import annotations + +import json +import subprocess +import unittest + +import openshift_campaign + + +class OpenShiftCampaignTests(unittest.TestCase): + def test_external_token_is_sent_over_stdin(self): + class FakeOc: + arguments = () + input_text = None + + def run(self, *arguments, input_text=None): + self.arguments = arguments + self.input_text = input_text + return subprocess.CompletedProcess( + arguments, 0, json.dumps(["echo"]) + "\n", "" + ) + + oc = FakeOc() + token = "private-token" + tools = openshift_campaign.configure_external( + oc, + "target-external", + "benchmark", + token, + "http://fast-time/mcp", + "2026-07-28", + ) + self.assertEqual(tools, ["echo"]) + self.assertEqual(oc.input_text, token + "\n") + self.assertNotIn(token, " ".join(oc.arguments)) + + def test_assigns_each_equal_lane_to_a_distinct_worker(self): + pools = [] + nodes = [] + for memory, prefix in ((12, "target"), (20, "locust"), (40, "fast-time")): + cpu = {12: 6, 20: 6, 40: 10}[memory] + for lane in ("builtin", "external"): + pools.append( + { + "role": f"{prefix}-{lane}", + "count": 1, + "cpu": cpu, + "memory_gb": memory, + } + ) + nodes.append( + { + "metadata": {"name": f"{prefix}-{lane}", "labels": {}}, + "status": { + "capacity": { + "cpu": str(cpu), + "memory": f"{memory * 1024 * 1024}Ki", + } + }, + } + ) + assigned = openshift_campaign.assign_nodes( + {"infrastructure": {"openshift": {"worker_pools": pools}}}, + {"items": list(reversed(nodes))}, + ) + self.assertEqual(set(assigned), {pool["role"] for pool in pools}) + self.assertEqual( + len({node for role_nodes in assigned.values() for node in role_nodes}), 6 + ) + + def test_balances_parallel_targets_across_repeated_role_workers(self): + nodes = {"target": ["target-a", "target-b"]} + selected = [ + openshift_campaign.assigned_node(nodes, "target", lane, users) + for users in (125, 250, 500, 1000) + for lane in ("builtin", "external") + ] + self.assertEqual(selected.count("target-a"), 4) + self.assertEqual(selected.count("target-b"), 4) + + def test_locust_pod_reserves_four_cpus_and_sixteen_gib(self): + config = { + "infrastructure": { + "openshift": { + "load_pod": { + "cpu_millicores": 4000, + "memory_mib": 16384, + } + } + }, + "images": {"locust": "locust@sha256:test"}, + "workload": { + "protocol_version": "2026-07-28", + "ramp_seconds": 30, + "warmup_seconds": 30, + "measure_seconds": 3600, + }, + } + pod = openshift_campaign.locust_pod( + config, + "benchmark", + "external", + "worker-1", + 125, + "http://target:4445", + ["echo"], + ) + containers = pod["spec"]["containers"] + cpu_milli = sum( + int(item["resources"]["limits"]["cpu"].removesuffix("m")) + if item["resources"]["limits"]["cpu"].endswith("m") + else 1000 * int(item["resources"]["limits"]["cpu"]) + for item in containers + ) + memory_mib = sum( + int(item["resources"]["limits"]["memory"].removesuffix("Mi")) + if item["resources"]["limits"]["memory"].endswith("Mi") + else 1024 * int(item["resources"]["limits"]["memory"].removesuffix("Gi")) + for item in containers + ) + self.assertEqual(cpu_milli, 4000) + self.assertEqual(memory_mib, 16384) + self.assertEqual(len(containers), 4) + + def test_memory_summary_adds_all_gateway_sidecars(self): + samples = [ + { + "pods": "target-builtin gateway 100m 1200Mi\n" + "builtin-db postgres 30m 400Mi\n" + "builtin-db redis 10m 100Mi\n" + "target-external dataplane 100m 900Mi\n" + }, + { + "pods": "target-builtin gateway 100m 1400Mi\n" + "builtin-db postgres 30m 450Mi\n" + "builtin-db redis 10m 110Mi\n" + }, + ] + memory = openshift_campaign.lane_memory(samples, "builtin") + self.assertEqual(memory["average_mib"], 1830) + self.assertEqual(memory["peak_mib"], 1960) + self.assertEqual(memory["samples"], 2) + + def test_memory_summary_excludes_warmup_samples(self): + samples = [ + {"time": 10, "pods": "target-external dataplane 100m 900Mi\n"}, + {"time": 20, "pods": "target-external dataplane 100m 400Mi\n"}, + {"time": 30, "pods": "target-external dataplane 100m 500Mi\n"}, + ] + memory = openshift_campaign.lane_memory( + samples, "external", start_time=20, end_time=30 + ) + self.assertEqual(memory["average_mib"], 450) + self.assertEqual(memory["peak_mib"], 500) + self.assertEqual(memory["samples"], 2) + + def test_helper_pressure_detects_sustained_locust_saturation(self): + samples = [ + { + "time": timestamp, + "pods": "load-external-125 master 100m 100Mi\n" + "load-external-125 worker-1 900m 100Mi\n" + "load-external-125 worker-2 900m 100Mi\n" + "load-external-125 worker-3 900m 100Mi\n" + "fast-time-external-125 fast-time 1000m 100Mi\n", + } + for timestamp in (20, 30, 40) + ] + pressure = openshift_campaign.helper_pressure( + samples, + "external", + { + "infrastructure": { + "openshift": { + "load_pod": { + "cpu_millicores": 4000, + "memory_mib": 16384, + }, + "backend_pod": { + "cpu_millicores": 8000, + "memory_mib": 32768, + }, + } + }, + "workload": { + "helper_cpu_percent": 70, + "helper_memory_percent": 80, + "worker_core_percent": 85, + } + }, + 125, + 20, + 40, + ) + self.assertIn("Locust worker CPU", pressure["saturated"]) + self.assertEqual(pressure["samples"], 3) + + +if __name__ == "__main__": + unittest.main() diff --git a/src/cli.rs b/src/cli.rs index 4ba9bed..a27a152 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -323,7 +323,7 @@ pub(crate) enum LoadCommand { /// Run Locust through the selected public MCP route. #[command(visible_alias = "r")] Run(LoadRunArgs), - /// Run repeatable comparison and scaling benchmarks on FYRE VMs. + /// Run repeatable comparison and scaling benchmarks on FYRE infrastructure. #[command(visible_alias = "f")] Fyre(FyreArgs), } @@ -339,13 +339,13 @@ pub(crate) struct FyreArgs { /// 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. + /// Provision, benchmark, download reports, and destroy run-owned infrastructure. #[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. + /// Destroy only the infrastructure owned by a benchmark run. #[command(visible_alias = "d")] Destroy(FyreExistingRunArgs), } diff --git a/src/infrastructure/assets.rs b/src/infrastructure/assets.rs index 843f4fa..15bec68 100644 --- a/src/infrastructure/assets.rs +++ b/src/infrastructure/assets.rs @@ -32,8 +32,11 @@ static ASSETS: LazyLock> = LazyLock::new(|| { asset!("Cargo.toml"), asset!("Cargo.lock"), asset!("benchmarks/fyre/scaling.yaml"), + asset!("benchmarks/fyre/openshift.yaml"), + asset!("benchmarks/fyre/openshift-2v2-parallel.yaml"), asset!("benchmarks/fyre/vertical-low-memory.yaml"), asset!("benchmarks/fyre/campaign.py"), + asset!("benchmarks/fyre/openshift_campaign.py"), asset!("benchmarks/fyre/report.py"), asset!("benchmarks/fyre/README.md"), asset!("benchmarks/fyre/deploy/dataplane.compose.yaml"), diff --git a/src/runtime/fyre.rs b/src/runtime/fyre.rs index a74c2e8..635a157 100644 --- a/src/runtime/fyre.rs +++ b/src/runtime/fyre.rs @@ -14,12 +14,18 @@ use uuid::Uuid; use super::{AppFailure, AppResult, CommandSpec, ProcessRunner, RuntimeContext}; use crate::app::FyreAction; +mod openshift; + const OWNERSHIP_FILE: &str = "run.json"; const TERRAFORM_DIRECTORY: &str = "terraform"; const TERRAFORM_VARIABLES: &str = "scenario.tfvars.json"; const HELPER_SATURATION_EXIT: i32 = 42; const FYRE_STANDALONE_UBUNTU_OS_DISK_GB: u32 = 250; +fn default_infrastructure_kind() -> String { + "standalone".to_owned() +} + #[derive(Debug, Clone, Serialize, Deserialize)] struct FyreConfig { schema_version: u32, @@ -37,16 +43,51 @@ struct FyreConfig { #[derive(Debug, Clone, Serialize, Deserialize)] struct InfrastructureConfig { + #[serde(default = "default_infrastructure_kind")] + kind: String, os: String, - ssh_user: String, - ssh_private_key: PathBuf, - ssh_public_key: PathBuf, + #[serde(default)] + ssh_user: Option, + #[serde(default)] + ssh_private_key: Option, + #[serde(default)] + ssh_public_key: Option, expiry_hours: u32, + #[serde(default)] + openshift: Option, + #[serde(default)] helper_sizes: Vec, #[serde(default)] initial_helpers: Option, } +#[derive(Debug, Clone, Serialize, Deserialize)] +struct OpenShiftConfig { + version: String, + base_disk_gb: u32, + oc_image: String, + master: MachineSize, + api: MachineSize, + target_pod: PodSize, + load_pod: PodSize, + backend_pod: PodSize, + worker_pools: Vec, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize)] +struct PodSize { + cpu_millicores: u32, + memory_mib: u32, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct OpenShiftWorkerPool { + role: String, + count: u32, + cpu: u32, + memory_gb: u32, +} + #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] struct MachineSize { cpu: u32, @@ -98,6 +139,10 @@ struct WorkloadConfig { helper_cpu_percent: f64, helper_memory_percent: f64, worker_core_percent: f64, + #[serde(default)] + parallel_lanes: bool, + #[serde(default)] + parallel_user_levels: bool, tools: Vec, } @@ -137,6 +182,10 @@ struct RequiredCapacity { struct RunState { schema_version: u32, run_id: String, + #[serde(default = "default_infrastructure_kind")] + infrastructure_kind: String, + #[serde(skip_serializing_if = "Option::is_none")] + cluster_name: Option, phase: String, config_file: PathBuf, current_scenario: Option, @@ -164,10 +213,25 @@ impl RuntimeContext { 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)?; + if config.infrastructure.kind == "openshift" { + return self.run_fyre_openshift(source, config, run_id).await; + } + let private_key = expand_home( + config + .infrastructure + .ssh_private_key + .as_ref() + .expect("validated standalone FYRE configuration has an SSH private key"), + ) + .map_err(AppFailure::from)?; + let public_key = expand_home( + config + .infrastructure + .ssh_public_key + .as_ref() + .expect("validated standalone FYRE configuration has an 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); @@ -197,6 +261,8 @@ impl RuntimeContext { let mut state = RunState { schema_version: 1, run_id: run_id.clone(), + infrastructure_kind: "standalone".to_owned(), + cluster_name: None, phase: "initializing".to_owned(), config_file: source, current_scenario: None, @@ -575,6 +641,11 @@ impl RuntimeContext { 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)?; + if state.infrastructure_kind == "openshift" { + state.phase = "destroying".to_owned(); + write_state(&root, &state).map_err(AppFailure::from)?; + return self.destroy_fyre_openshift(&root, &mut state).await; + } let terraform = terraform_binary().map_err(AppFailure::from)?; state.phase = "destroying".to_owned(); write_state(&root, &state).map_err(AppFailure::from)?; @@ -707,9 +778,30 @@ fn validate_config(config: &FyreConfig) -> Result<()> { "unsupported FYRE configuration schema" ); ensure!( - config.infrastructure.os == "Ubuntu 24.04", - "FYRE benchmark OS must be Ubuntu 24.04" + matches!( + config.infrastructure.kind.as_str(), + "standalone" | "openshift" + ), + "FYRE infrastructure kind must be standalone or openshift" ); + if config.infrastructure.kind == "standalone" { + ensure!( + config.infrastructure.os == "Ubuntu 24.04", + "standalone FYRE benchmark OS must be Ubuntu 24.04" + ); + ensure!( + config + .infrastructure + .ssh_user + .as_deref() + .is_some_and(|value| !value.is_empty()) + && config.infrastructure.ssh_private_key.is_some() + && config.infrastructure.ssh_public_key.is_some(), + "standalone FYRE infrastructure requires SSH user and key paths" + ); + } else { + validate_openshift_config(config)?; + } ensure!( (8..=24).contains(&config.infrastructure.expiry_hours), "FYRE expiry must be between eight and 24 hours" @@ -750,7 +842,7 @@ fn validate_config(config: &FyreConfig) -> Result<()> { ); ensure!( config.scenarios.len() == 1 && config.scenarios[0].replicas == 1, - "the comparison target must be exactly one VM" + "the comparison target must be exactly one isolated allocation" ); ensure!( config.images.controlplane.is_some() && config.images.postgres.is_some(), @@ -777,20 +869,22 @@ fn validate_config(config: &FyreConfig) -> Result<()> { 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" - ); - initial_helper_indices(config)?; - 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" - ); + if config.infrastructure.kind == "standalone" { + ensure!( + !config.infrastructure.helper_sizes.is_empty(), + "at least one helper size is required" + ); + initial_helper_indices(config)?; + 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(); let baseline = config .scenarios @@ -840,6 +934,9 @@ fn validate_config(config: &FyreConfig) -> Result<()> { ]; images.extend(config.images.controlplane.iter()); images.extend(config.images.postgres.iter()); + if let Some(openshift) = &config.infrastructure.openshift { + images.push(&openshift.oc_image); + } for image in images { ensure!( image.contains("@sha256:"), @@ -849,6 +946,115 @@ fn validate_config(config: &FyreConfig) -> Result<()> { Ok(()) } +fn validate_openshift_config(config: &FyreConfig) -> Result<()> { + let openshift = config + .infrastructure + .openshift + .as_ref() + .context("OpenShift infrastructure settings are required")?; + ensure!( + config.benchmark_kind == "comparison", + "OpenShift currently supports only the built-in/external comparison" + ); + ensure!( + config.workload.parallel_lanes, + "OpenShift must run the built-in and external lanes in parallel" + ); + ensure!( + openshift.base_disk_gb == 40, + "OpenShift benchmark nodes must use the 40 GB FYRE minimum disk" + ); + ensure!( + !openshift.version.trim().is_empty(), + "OpenShift version must not be empty" + ); + ensure!( + openshift.master.cpu > 0 + && openshift.master.memory_gb > 0 + && openshift.api.cpu > 0 + && openshift.api.memory_gb > 0, + "OpenShift control-plane resources must be positive" + ); + let expected = if config.workload.parallel_user_levels { + BTreeSet::from(["target", "locust", "fast-time"]) + } else { + BTreeSet::from([ + "target-builtin", + "target-external", + "locust-builtin", + "locust-external", + "fast-time-builtin", + "fast-time-external", + ]) + }; + let mut roles = BTreeSet::new(); + for pool in &openshift.worker_pools { + ensure!( + pool.count > 0 && pool.cpu > 0 && pool.cpu <= 16 && pool.memory_gb > 0, + "OpenShift worker pool {} must contain positive-sized workers with at most 16 vCPU each", + pool.role + ); + ensure!( + roles.insert(pool.role.as_str()), + "duplicate OpenShift worker role {}", + pool.role + ); + } + ensure!( + roles == expected, + "OpenShift worker roles do not match the selected parallel topology" + ); + ensure!( + openshift.load_pod.cpu_millicores > 0 + && openshift.load_pod.memory_mib > 0 + && openshift.target_pod.cpu_millicores > 0 + && openshift.target_pod.memory_mib > 0 + && openshift.backend_pod.cpu_millicores > 0 + && openshift.backend_pod.memory_mib > 0, + "OpenShift helper pod resources must be positive" + ); + if config.workload.parallel_user_levels { + let measurements = (config.workload.user_levels.len() * 2) as u32; + let pool = |role: &str| { + openshift + .worker_pools + .iter() + .find(|pool| pool.role == role) + .expect("validated OpenShift worker role") + }; + let target = pool("target"); + let scenario = &config.scenarios[0]; + ensure!( + openshift.target_pod.cpu_millicores <= scenario.cpu * 1_000 + && openshift.target_pod.memory_mib <= scenario.memory_gb * 1_024, + "OpenShift target requests cannot exceed the target limits" + ); + ensure!( + openshift.target_pod.cpu_millicores * measurements <= target.cpu * target.count * 1_000 + && openshift.target_pod.memory_mib * measurements + <= target.memory_gb * target.count * 1_024, + "shared OpenShift target workers cannot reserve every parallel target" + ); + for (role, pod) in [ + ("locust", openshift.load_pod), + ("fast-time", openshift.backend_pod), + ] { + let worker = pool(role); + ensure!( + pod.cpu_millicores * measurements <= worker.cpu * worker.count * 1_000 + && pod.memory_mib * measurements <= worker.memory_gb * worker.count * 1_024, + "shared OpenShift {role} worker cannot reserve every parallel benchmark pod" + ); + } + } else { + ensure!( + openshift.worker_pools.iter().all(|pool| pool.count == 1), + "isolated OpenShift lane worker pools must contain exactly one worker" + ); + } + Ok(()) +} + fn initial_helper_indices(config: &FyreConfig) -> Result<(usize, usize)> { let Some(initial) = config.infrastructure.initial_helpers else { return Ok((0, 0)); @@ -1064,10 +1270,17 @@ fn read_owned_state(root: &Path, expected_run_id: &str) -> Result { 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" - ); + if state.infrastructure_kind == "standalone" { + ensure!( + root.join(TERRAFORM_DIRECTORY).is_dir(), + "FYRE Terraform state directory is missing; refusing cleanup" + ); + } else { + ensure!( + state.infrastructure_kind == "openshift" && state.cluster_name.is_some(), + "FYRE OpenShift ownership metadata is missing; refusing cleanup" + ); + } Ok(state) } @@ -1137,6 +1350,8 @@ mod tests { let state = RunState { schema_version: 1, run_id: "owned-run".to_owned(), + infrastructure_kind: "standalone".to_owned(), + cluster_name: None, phase: "failed".to_owned(), config_file: PathBuf::from("config.yaml"), current_scenario: None, diff --git a/src/runtime/fyre/openshift.rs b/src/runtime/fyre/openshift.rs new file mode 100644 index 0000000..db29fc8 --- /dev/null +++ b/src/runtime/fyre/openshift.rs @@ -0,0 +1,703 @@ +//! FYRE OpenShift lifecycle for the isolated parallel comparison. + +use std::collections::BTreeMap; +use std::fs; +use std::path::{Path, PathBuf}; +use std::time::Duration; + +use anyhow::{Context, Result, bail, ensure}; +use reqwest::{Client, Response, StatusCode}; +use serde_json::{Value, json}; +use url::Url; + +use super::{FyreConfig, OpenShiftConfig, RunState, validate_run_id, write_json, write_state}; +use crate::infrastructure::process::CommandSpec; +use crate::runtime::{AppFailure, AppResult, ProcessRunner, RuntimeContext}; + +const DEFAULT_API_BASE: &str = "https://ocpapi.svl.ibm.com/v1"; +const PROVISION_POLL: Duration = Duration::from_secs(120); +const PROVISION_TIMEOUT: Duration = Duration::from_secs(7_200); +const DELETE_POLL: Duration = Duration::from_secs(30); +const DELETE_TIMEOUT: Duration = Duration::from_secs(1_800); + +struct FyreOpenShiftApi { + http: Client, + base: Url, + username: String, + api_key: String, + site: String, + product_group_id: String, +} + +impl FyreOpenShiftApi { + fn new(runtime: &RuntimeContext) -> Result { + let username = runtime + .fyre_text("FYRE_USERNAME") + .context("FYRE_USERNAME is required for FYRE provisioning")? + .to_owned(); + let api_key = runtime + .fyre_text("FYRE_API_KEY") + .context("FYRE_API_KEY is required for FYRE provisioning")? + .to_owned(); + let site = runtime.fyre_text("FYRE_SITE").unwrap_or("svl").to_owned(); + ensure!( + matches!(site.as_str(), "svl" | "rtp"), + "FYRE_SITE must be svl or rtp" + ); + let product_group_id = runtime + .fyre_text("FYRE_PRODUCT_GROUP_ID") + .context("FYRE_PRODUCT_GROUP_ID is required for 40 GB OpenShift nodes")? + .to_owned(); + let base = runtime + .fyre_text("FYRE_OCP_API_URL") + .unwrap_or(DEFAULT_API_BASE) + .trim_end_matches('/'); + let base = + Url::parse(&format!("{base}/")).context("FYRE_OCP_API_URL must be an absolute URL")?; + let http = Client::builder() + .danger_accept_invalid_certs(true) + .connect_timeout(Duration::from_secs(30)) + .timeout(Duration::from_secs(300)) + .redirect(reqwest::redirect::Policy::none()) + .build() + .context("failed to create FYRE OpenShift API client")?; + Ok(Self { + http, + base, + username, + api_key, + site, + product_group_id, + }) + } + + fn request(&self, method: reqwest::Method, path: &str) -> reqwest::RequestBuilder { + let mut url = self + .base + .join(path.trim_start_matches('/')) + .expect("validated FYRE API base must accept relative paths"); + url.query_pairs_mut().append_pair("site", &self.site); + self.http + .request(method, url) + .basic_auth(&self.username, Some(&self.api_key)) + } + + async fn hostname_available(&self, cluster: &str) -> Result { + let response = self + .request(reqwest::Method::GET, &format!("/check_hostname/{cluster}")) + .send() + .await + .context("failed to check FYRE OpenShift cluster name")?; + let value = response_json(response, "check FYRE OpenShift cluster name").await?; + Ok(value["status"] == "success") + } + + async fn create(&self, cluster: &str, config: &FyreConfig) -> Result<()> { + ensure!( + self.hostname_available(cluster).await?, + "FYRE OpenShift cluster {cluster} already exists" + ); + let openshift = config + .infrastructure + .openshift + .as_ref() + .context("OpenShift settings are missing")?; + let payload = cluster_payload( + cluster, + config, + openshift, + &self.product_group_id, + &self.site, + ); + for attempt in 0..10 { + let response = self + .request(reqwest::Method::POST, "/ocp/x") + .json(&payload) + .send() + .await + .context("failed to request FYRE OpenShift cluster")?; + if response.status() == StatusCode::TOO_MANY_REQUESTS && attempt < 9 { + tokio::time::sleep(PROVISION_POLL).await; + continue; + } + response_json(response, "create FYRE OpenShift cluster").await?; + return Ok(()); + } + bail!("FYRE OpenShift provisioning remained rate limited after ten attempts") + } + + async fn wait_deployed(&self, cluster: &str) -> Result { + let operation = async { + loop { + let response = self + .request(reqwest::Method::GET, &format!("/ocp/{cluster}/status")) + .send() + .await + .context("failed to read FYRE OpenShift deployment status")?; + let (http_status, value) = + decode_response(response, "read FYRE OpenShift deployment status").await?; + let details = value["details"].as_str().unwrap_or_default(); + if http_status == StatusCode::BAD_REQUEST && details.contains("does not exist") { + tokio::time::sleep(PROVISION_POLL).await; + continue; + } + ensure_success(http_status, &value, "read FYRE OpenShift deployment status")?; + let status = value["deployed_status"] + .as_str() + .or_else(|| value["cluster_status"].as_str()) + .or_else(|| value["status"].as_str()) + .unwrap_or("unknown"); + if status == "deployed" { + return self.details(cluster).await; + } + if matches!(status, "failed" | "error" | "deleted") { + bail!("FYRE OpenShift deployment entered {status} state") + } + tokio::time::sleep(PROVISION_POLL).await; + } + }; + tokio::time::timeout(PROVISION_TIMEOUT, operation) + .await + .context("FYRE OpenShift cluster did not deploy within two hours")? + } + + async fn details(&self, cluster: &str) -> Result { + let response = self + .request(reqwest::Method::GET, &format!("/ocp/{cluster}")) + .send() + .await + .context("failed to read FYRE OpenShift cluster details")?; + response_json(response, "read FYRE OpenShift cluster details").await + } + + async fn delete(&self, cluster: &str) -> Result<()> { + if self.hostname_available(cluster).await? { + return Ok(()); + } + let response = self + .request(reqwest::Method::DELETE, &format!("/ocp/{cluster}")) + .send() + .await + .context("failed to delete FYRE OpenShift cluster")?; + response_json(response, "delete FYRE OpenShift cluster").await?; + tokio::time::timeout(DELETE_TIMEOUT, async { + loop { + if self.hostname_available(cluster).await? { + return Ok(()); + } + tokio::time::sleep(DELETE_POLL).await; + } + }) + .await + .context("FYRE OpenShift cluster deletion did not finish within 30 minutes")? + } +} + +impl RuntimeContext { + pub(super) async fn run_fyre_openshift( + &self, + source: PathBuf, + config: FyreConfig, + run_id: Option, + ) -> AppResult<()> { + let run_id = run_id + .unwrap_or_else(|| format!("ocp-{}", &uuid::Uuid::new_v4().simple().to_string()[..12])); + validate_run_id(&run_id).map_err(AppFailure::from)?; + let cluster_name = format!("cf-{run_id}"); + if cluster_name.len() > 32 { + return Err(AppFailure::from(anyhow::anyhow!( + "OpenShift run ID must be at most 29 characters" + ))); + } + let root = self.config.integration_dir().join("fyre").join(&run_id); + let config_path = root.join("config.json"); + let resuming = root.exists(); + let (config, mut state) = if resuming { + let state = super::read_owned_state(&root, &run_id).map_err(AppFailure::from)?; + if !state.cleanup_required { + return Err(AppFailure::from(anyhow::anyhow!( + "FYRE OpenShift run {run_id} is already {}", + state.phase + ))); + } + if state.cluster_name.as_deref() != Some(cluster_name.as_str()) { + return Err(AppFailure::from(anyhow::anyhow!( + "FYRE OpenShift run {run_id} owns a different cluster; refusing resume" + ))); + } + let saved: FyreConfig = serde_json::from_slice( + &fs::read(&config_path) + .with_context(|| format!("failed to read {}", config_path.display())) + .map_err(AppFailure::from)?, + ) + .context("invalid saved FYRE OpenShift configuration") + .map_err(AppFailure::from)?; + println!( + "Resuming FYRE OpenShift run {run_id} from phase {}", + state.phase + ); + (saved, state) + } else { + fs::create_dir_all(root.join("results/comparison")) + .with_context(|| format!("failed to create FYRE run directory {}", root.display())) + .map_err(AppFailure::from)?; + write_json(&config_path, &config).map_err(AppFailure::from)?; + let state = RunState { + schema_version: 1, + run_id: run_id.clone(), + infrastructure_kind: "openshift".to_owned(), + cluster_name: Some(cluster_name.clone()), + phase: "provisioning".to_owned(), + config_file: source, + current_scenario: Some("comparison".to_owned()), + locust_helper_size: 0, + fast_time_helper_size: 0, + completed_scenarios: Vec::new(), + cleanup_required: true, + }; + write_state(&root, &state).map_err(AppFailure::from)?; + (config, state) + }; + let api = FyreOpenShiftApi::new(self).map_err(AppFailure::from)?; + + let primary = async { + let provision = async { + if resuming { + ensure!( + !api.hostname_available(&cluster_name).await?, + "FYRE OpenShift cluster {cluster_name} no longer exists; use destroy to close the stale run" + ); + } else { + api.create(&cluster_name, &config).await?; + } + api.wait_deployed(&cluster_name).await + }; + tokio::pin!(provision); + let details = tokio::select! { + result = &mut provision => result.map_err(AppFailure::from)?, + signal = tokio::signal::ctrl_c() => { + signal + .context("failed to install interrupt handler") + .map_err(AppFailure::from)?; + return Err(AppFailure::from(anyhow::anyhow!( + "FYRE OpenShift provisioning interrupted" + ))); + } + }; + write_redacted_details(&root.join("cluster.json"), &details) + .map_err(AppFailure::from)?; + state.phase = "authenticating".to_owned(); + write_state(&root, &state).map_err(AppFailure::from)?; + self.openshift_login(&root, &cluster_name, &config, &details) + .await?; + let scenario_root = root.join("results/comparison"); + if !state.completed_scenarios.iter().any(|item| item == "comparison") { + state.phase = "benchmarking".to_owned(); + write_state(&root, &state).map_err(AppFailure::from)?; + let command = CommandSpec::new("python3") + .arg( + self.config + .asset_root() + .join("benchmarks/fyre/openshift_campaign.py"), + ) + .arg("--config") + .arg(&config_path) + .arg("--kubeconfig") + .arg(root.join("kubeconfig")) + .arg("--output") + .arg(&scenario_root) + .arg("--run-id") + .arg(&run_id) + .arg("--assets") + .arg(self.config.asset_root().join("benchmarks/fyre")); + self.run_cancellable(&command).await?; + state.completed_scenarios.push("comparison".to_owned()); + state.current_scenario = None; + write_state(&root, &state).map_err(AppFailure::from)?; + } + self.generate_fyre_report(&root, &config_path).await?; + write_json( + &root.join("manifest.json"), + &json!({ + "schema_version": 2, + "run_id": run_id, + "cluster_name": cluster_name, + "configuration": config, + "state": state, + "cluster": redacted_details(&details), + }), + ) + .map_err(AppFailure::from) + } + .await; + + state.phase = "destroying".to_owned(); + let _ = write_state(&root, &state); + let cleanup = retry_delete(&api, &cluster_name) + .await + .map_err(AppFailure::from); + let _ = fs::remove_file(root.join("kubeconfig")); + 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::super::finish_with_cleanup(primary.err(), cleanup) + } + + pub(super) async fn destroy_fyre_openshift( + &self, + root: &Path, + state: &mut RunState, + ) -> AppResult<()> { + let cluster = state.cluster_name.as_deref().ok_or_else(|| { + AppFailure::from(anyhow::anyhow!( + "FYRE OpenShift cluster ownership is missing" + )) + })?; + let api = FyreOpenShiftApi::new(self).map_err(AppFailure::from)?; + retry_delete(&api, cluster) + .await + .map_err(AppFailure::from)?; + state.cleanup_required = false; + state.phase = "destroyed".to_owned(); + write_state(root, state).map_err(AppFailure::from) + } + + async fn openshift_login( + &self, + root: &Path, + cluster: &str, + config: &FyreConfig, + details: &Value, + ) -> AppResult<()> { + let password = cluster_record(details, cluster) + .and_then(|record| record["kubeadmin_password"].as_str()) + .filter(|value| !value.is_empty()) + .context("FYRE cluster details did not include kubeadmin_password") + .map_err(AppFailure::from)?; + let image = &config + .infrastructure + .openshift + .as_ref() + .expect("validated OpenShift configuration") + .oc_image; + let mount = format!("{}:/work", root.display()); + let command = CommandSpec::new("docker") + .args([ + "run", + "--rm", + "--platform", + "linux/amd64", + "--entrypoint", + "/bin/sh", + ]) + .arg("-v") + .arg(mount) + .arg("-e") + .arg("OCP_PASSWORD") + .env("OCP_PASSWORD", password) + .arg(image) + .args([ + "-ceu", + "for attempt in 1 2 3 4 5 6 7 8 9 10; do oc login -u kubeadmin -p \"$OCP_PASSWORD\" --server \"$1\" --insecure-skip-tls-verify=true --kubeconfig /work/kubeconfig && exit 0; [ \"$attempt\" = 10 ] || sleep 30; done; exit 1", + "oc-login", + ]) + .arg(format!("https://api.{cluster}.cp.fyre.ibm.com:6443")); + self.run_cancellable(&command).await?; + set_private_permissions(&root.join("kubeconfig")).map_err(AppFailure::from) + } +} + +fn cluster_payload( + cluster: &str, + config: &FyreConfig, + openshift: &OpenShiftConfig, + product_group: &str, + site: &str, +) -> Value { + let base_disk_size = openshift.base_disk_gb.to_string(); + let mut worker_pools = BTreeMap::<(u32, u32), u32>::new(); + for pool in &openshift.worker_pools { + *worker_pools.entry((pool.cpu, pool.memory_gb)).or_default() += pool.count; + } + json!({ + "name": cluster, + "description": "ContextForge parallel built-in/external dataplane benchmark", + "platform": "x", + "quota_type": "product_group", + "site": site, + "product_group_id": product_group, + "ocp_version": openshift.version, + "expiration": format!("{} hours", config.infrastructure.expiry_hours), + "ipv6_test": false, + "fips": "no", + "master": { + "count": 3, + "cpu": openshift.master.cpu, + "memory": openshift.master.memory_gb, + "disk": openshift.base_disk_gb, + "base_disk_size": base_disk_size, + }, + "api": { + "count": 1, + "cpu": openshift.api.cpu, + "memory": openshift.api.memory_gb, + "disk": openshift.base_disk_gb, + "base_disk_size": base_disk_size, + }, + "infra": { + "cpu": openshift.api.cpu, + "memory": openshift.api.memory_gb, + "disk": openshift.base_disk_gb, + }, + "worker": worker_pools.into_iter().map(|((cpu, memory), count)| json!({ + "count": count, + "cpu": cpu, + "memory": memory, + "os_disk": openshift.base_disk_gb, + "base_disk_size": base_disk_size, + "additional_disk": [], + })).collect::>(), + }) +} + +async fn response_json(response: Response, operation: &str) -> Result { + let (status, value) = decode_response(response, operation).await?; + ensure_success(status, &value, operation)?; + Ok(value) +} + +async fn decode_response(response: Response, operation: &str) -> Result<(StatusCode, Value)> { + let status = response.status(); + let bytes = response + .bytes() + .await + .with_context(|| format!("failed to read response for {operation}"))?; + let value = serde_json::from_slice::(&bytes).unwrap_or_else( + |_| json!({"details": String::from_utf8_lossy(&bytes[..bytes.len().min(2_048)])}), + ); + Ok((status, value)) +} + +fn ensure_success(status: StatusCode, value: &Value, operation: &str) -> Result<()> { + let details = value + .get("details") + .filter(|details| !details.is_null()) + .map(|details| { + details + .as_str() + .map(str::to_owned) + .unwrap_or_else(|| details.to_string()) + }) + .unwrap_or_else(|| value.to_string()); + ensure!( + status.is_success(), + "{operation} failed with HTTP {status}: {details}" + ); + Ok(()) +} + +fn cluster_record<'a>(details: &'a Value, cluster: &str) -> Option<&'a Value> { + details["clusters"] + .as_array()? + .iter() + .find(|record| record["cluster_name"] == cluster || record["name"] == cluster) +} + +fn redacted_details(details: &Value) -> Value { + fn redact(value: &mut Value) { + match value { + Value::Object(object) => { + object.retain(|key, _| { + let key = key.to_ascii_lowercase(); + !["password", "token", "secret", "api_key", "kubeconfig"] + .iter() + .any(|sensitive| key.contains(sensitive)) + }); + for child in object.values_mut() { + redact(child); + } + } + Value::Array(values) => { + for child in values { + redact(child); + } + } + _ => {} + } + } + let mut redacted = details.clone(); + redact(&mut redacted); + redacted +} + +fn write_redacted_details(path: &Path, details: &Value) -> Result<()> { + write_json(path, &redacted_details(details)) +} + +async fn retry_delete(api: &FyreOpenShiftApi, cluster: &str) -> Result<()> { + let mut last = None; + for attempt in 0..3 { + match api.delete(cluster).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(|| anyhow::anyhow!("FYRE OpenShift deletion failed"))) +} + +#[cfg(unix)] +fn set_private_permissions(path: &Path) -> Result<()> { + use std::os::unix::fs::PermissionsExt; + fs::set_permissions(path, fs::Permissions::from_mode(0o600)) + .with_context(|| format!("failed to secure {}", path.display())) +} + +#[cfg(not(unix))] +fn set_private_permissions(path: &Path) -> Result<()> { + ensure!( + path.is_file(), + "kubeconfig {} was not created", + path.display() + ); + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn redaction_removes_cluster_credentials() { + let value = json!({"clusters": [{ + "cluster_name": "cf-run", + "kubeadmin_password": "secret", + "token": "secret", + "vms": [{"nested_secret": "secret", "address": "private"}] + }]}); + let redacted = redacted_details(&value); + assert!(redacted["clusters"][0].get("kubeadmin_password").is_none()); + assert!(redacted["clusters"][0].get("token").is_none()); + assert!( + redacted["clusters"][0]["vms"][0] + .get("nested_secret") + .is_none() + ); + assert_eq!(redacted["clusters"][0]["vms"][0]["address"], "private"); + assert_eq!(redacted["clusters"][0]["cluster_name"], "cf-run"); + } + + #[test] + fn packaged_profile_uses_three_pairs_of_40_gb_workers() { + let config = super::super::read_config( + Path::new(env!("CARGO_MANIFEST_DIR")) + .join("benchmarks/fyre/openshift.yaml") + .as_path(), + ) + .expect("packaged OpenShift profile"); + super::super::validate_config(&config).expect("valid OpenShift profile"); + let openshift = config + .infrastructure + .openshift + .as_ref() + .expect("OpenShift settings"); + let payload = cluster_payload("cf-test", &config, openshift, "808", "svl"); + assert_eq!(payload["master"]["disk"], 40); + assert_eq!(payload["master"]["base_disk_size"], "40"); + let workers = payload["worker"].as_array().expect("worker pools"); + assert_eq!(workers.len(), 3); + assert!(workers.iter().all(|pool| pool["count"] == 2)); + assert!(workers.iter().all(|pool| pool["os_disk"] == 40)); + assert!(workers.iter().all(|pool| pool["base_disk_size"] == "40")); + assert!( + workers + .iter() + .all(|pool| pool["additional_disk"] == json!([])) + ); + } + + #[test] + fn parallel_2v2_profile_fits_three_40_gb_workers() { + let config = super::super::read_config( + Path::new(env!("CARGO_MANIFEST_DIR")) + .join("benchmarks/fyre/openshift-2v2-parallel.yaml") + .as_path(), + ) + .expect("packaged parallel 2v2 OpenShift profile"); + super::super::validate_config(&config).expect("valid parallel 2v2 profile"); + let openshift = config + .infrastructure + .openshift + .as_ref() + .expect("OpenShift settings"); + let payload = cluster_payload("cf-test", &config, openshift, "808", "svl"); + let workers = payload["worker"].as_array().expect("worker pools"); + assert_eq!(payload["master"]["count"], 3); + assert_eq!(payload["master"]["base_disk_size"], "40"); + assert_eq!(payload["api"]["base_disk_size"], "40"); + assert_eq!(payload["infra"]["disk"], 40); + assert_eq!(workers.len(), 2); + assert_eq!( + workers + .iter() + .map(|pool| pool["count"].as_u64().expect("worker count")) + .sum::(), + 3 + ); + assert!(workers.iter().all(|pool| pool["os_disk"] == 40)); + assert!(workers.iter().all(|pool| pool["base_disk_size"] == "40")); + assert!( + workers + .iter() + .all(|pool| pool["additional_disk"] == json!([])) + ); + let worker_cpu: u64 = workers + .iter() + .map(|pool| { + pool["count"].as_u64().expect("worker count") + * pool["cpu"].as_u64().expect("worker cpu") + }) + .sum(); + let worker_memory: u64 = workers + .iter() + .map(|pool| { + pool["count"].as_u64().expect("worker count") + * pool["memory"].as_u64().expect("worker memory") + }) + .sum(); + let worker_disk: u64 = workers + .iter() + .map(|pool| { + pool["count"].as_u64().expect("worker count") + * pool["os_disk"].as_u64().expect("worker disk") + }) + .sum(); + assert_eq!(worker_cpu + 3 * 4 + 4, 60); + assert_eq!(worker_memory + 3 * 16 + 8, 96); + // FYRE fixes the API VM at 500 GB. Empty additional_disk arrays prevent + // the default two 200 GB data disks from being attached to every worker. + assert_eq!(worker_disk + 3 * 40 + 500, 740); + } + + #[test] + fn nested_api_errors_remain_actionable() { + let error = ensure_success( + StatusCode::BAD_REQUEST, + &json!({"status": "error", "details": {"errors": ["disk quota exceeded"]}}), + "create cluster", + ) + .expect_err("bad request must fail"); + assert!(error.to_string().contains("disk quota exceeded")); + } +}