diff --git a/benchmarking/README.md b/benchmarking/README.md index 7251e3c383..eca204b4cd 100644 --- a/benchmarking/README.md +++ b/benchmarking/README.md @@ -79,6 +79,13 @@ not a local entry point. See [automation/README.md](automation/README.md). python3 runner.py -f tests/.py -t 1m -u 1 --name --dest /tmp/bench ``` +One flag controls the optional post-run measurements described in +[Benchmark output files](#benchmark-output-files): + +* `--cluster-facts` / `--no-cluster-facts`: read node capacity and worker pod + count from the Kubernetes API once the run ends, to derive density frontiers. + On by default. Pass `--no-cluster-facts` to skip Kubernetes API discovery. + Test-specific flags are appended to the same command; see the sections below. ### DurDir Benchmark @@ -114,6 +121,52 @@ You must have enabled otel tracing for your cluster to view traces. You can find trace IDs by viewing the `logs` tab in the Locust UI +## Benchmark output files + +A run writes the following to `--dest`. Each run produces them fresh; none of +them are checked into the repository. + +* `status.json`: `locust_exit_code` and `stats_generated`. Deliberately just + those two keys, because it is what CI orchestration reads to decide whether a + trial ran at all. +* `stats.csv`, `stats_history.csv`, `failures.csv`, `exceptions.csv`: Locust's + own CSV output. +* `logs.txt`, `traces.txt`: the runner log, and the trace IDs seen during the run. +* `stats.jsonl`: one JSON object per line, one per metric. Every row carries + the same five keys: `timestamp`, `tag`, `test_name`, `metric`, and a flat + `measurements` map holding that metric's numbers. + +### Density frontiers + +With cluster discovery enabled, `stats.jsonl` gains a `trial_summary` row +describing how densely actors packed onto the hardware. Its `measurements` +map holds the raw facts and the derived numbers side by side. + +* `machine_type`, `node_count`, `allocatable_cores`, `allocatable_ram_gb` + (GiB), `worker_pod_count`: the measured facts, before any arithmetic. + Capacity covers the nodes the worker pods are running on rather than the + whole cluster, so a separate infrastructure pool is not counted. They are + recorded so the ratios below can be re-derived later, or recomputed against + a different denominator. +* `actors_per_node`, `actors_per_vcpu`, `actors_per_gb_ram`: the most users + Locust reported running, over the matching capacity. The `-u` flag only + stands in when no sample was read. +* `actors_per_pod_p50`, `actors_per_pod_p90`, `actors_per_pod_p99`: users per + worker pod across the run. Reported as a distribution rather than one + average, and it spans ramp-up too, because a custom load shape has no + single user count to call steady. +* `aggregate_failure_ratio`: failures over requests for the run. +* `_failure_ratio`: the same ratio for every operation Locust + reported, so each test carries its own names through. The operation name is + lowercased with underscores, so `DurDirWrite` becomes + `dur_dir_write_failure_ratio`. A key is absent when the test has no such + row, and null when the row ran no requests. + + +The Kubernetes API is not required. If it is unreachable, or discovery was +skipped, the affected fields are written as `null` and the run still +succeeds. A `null` means the value was not measured. It never means zero. + ## Optional: Prometheus + Grafana Locust provides graphs, statistics, etc. via the UI. However, you @@ -138,3 +191,12 @@ Once installed: code; it manages its own virtual environment under `locust/codegen/venv`. `hack/verify/codegen.sh` fails if the checked-in clients have drifted from the protos. + +### Unit tests + +`locust/unit_tests` covers the runner's helpers and needs no cluster. From the +repository root: + +```bash +python3 -m unittest discover -s benchmarking/locust/unit_tests +``` diff --git a/benchmarking/automation/manifests/runner-job.yaml.tmpl b/benchmarking/automation/manifests/runner-job.yaml.tmpl index 5855d1d05a..c43d678261 100644 --- a/benchmarking/automation/manifests/runner-job.yaml.tmpl +++ b/benchmarking/automation/manifests/runner-job.yaml.tmpl @@ -41,6 +41,53 @@ roleRef: name: atelet-endpointslices apiGroup: rbac.authorization.k8s.io --- +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRole +metadata: + name: benchmark-runner-hardware-discovery +rules: +- apiGroups: [""] + resources: ["nodes"] + # list only: the runner reads whole collections, never a single object. + verbs: ["list"] +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRoleBinding +metadata: + name: benchmark-runner-hardware-discovery +roleRef: + apiGroup: rbac.authorization.k8s.io + kind: ClusterRole + name: benchmark-runner-hardware-discovery +subjects: +- kind: ServiceAccount + name: benchmark-runner + namespace: benchmarking +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: Role +metadata: + name: benchmark-runner-worker-pods + namespace: benchmark-workloads +rules: +- apiGroups: [""] + resources: ["pods"] + verbs: ["list"] +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: RoleBinding +metadata: + name: benchmark-runner-worker-pods + namespace: benchmark-workloads +roleRef: + apiGroup: rbac.authorization.k8s.io + kind: Role + name: benchmark-runner-worker-pods +subjects: +- kind: ServiceAccount + name: benchmark-runner + namespace: benchmarking +--- apiVersion: batch/v1 kind: Job metadata: diff --git a/benchmarking/locust/Dockerfile b/benchmarking/locust/Dockerfile index 6de127ec1d..90644e315b 100644 --- a/benchmarking/locust/Dockerfile +++ b/benchmarking/locust/Dockerfile @@ -44,6 +44,7 @@ COPY benchmarking/locust/common/ /app/common/ COPY benchmarking/locust/shapes/ /app/shapes/ COPY benchmarking/locust/tests/ /app/tests/ COPY benchmarking/locust/runner.py /app/runner.py +COPY benchmarking/locust/cluster_facts.py /app/cluster_facts.py ENV PYTHONPATH=/app:/app/deps ENV PYTHONUNBUFFERED=1 diff --git a/benchmarking/locust/cluster_facts.py b/benchmarking/locust/cluster_facts.py new file mode 100644 index 0000000000..731232c3e8 --- /dev/null +++ b/benchmarking/locust/cluster_facts.py @@ -0,0 +1,266 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + + +"""Discovers cluster hardware capacity and records per-trial density frontiers. + +Reads allocatable CPU/RAM, node count and worker pod count from the Kubernetes +API, then derives the actor-density frontiers (actors per node / vCPU / GB RAM +and the actors-per-pod percentiles) for a completed trial. +""" + +import argparse +import csv +import json +import re +from pathlib import Path +from typing import Any, TextIO + +from kubernetes import client, config +from kubernetes.utils import parse_quantity + +API_TIMEOUT_SECONDS = 5 +WORKER_POOL_NAMESPACE = "benchmark-workloads" +WORKER_POOL_LABEL = "ate.dev/worker-pool" +LIVE_POD_PHASES = ("Running", "Pending") +MACHINE_TYPE_LABEL = "node.kubernetes.io/instance-type" + +# Shape returned when the cluster cannot be read, or when discovery is skipped +# with --no-cluster-facts. Keeping one definition means a trial_summary row has +# the same hardware keys either way, so consumers never have to special-case it. +EMPTY_FACTS: dict[str, Any] = { + "machine_type": None, + "node_count": None, + "allocatable_cores": None, + "allocatable_ram_gb": None, + "worker_pod_count": None, +} + + +def _log(logs: TextIO | None, msg: str) -> None: + """Mirrors runner.tee without importing it, to avoid a circular import.""" + print(msg, flush=True) + if logs is not None: + logs.write(msg + "\n") + logs.flush() + + +def _load_kube_config(logs: TextIO | None = None) -> bool: + """Loads in-cluster credentials, falling back to a local kubeconfig.""" + try: + config.load_incluster_config() + return True + except config.ConfigException: + pass + try: + config.load_kube_config() + return True + except config.ConfigException as e: + _log(logs, f"Notice: no Kubernetes credentials available: {e}") + return False + + +def _list_worker_pods( + v1: client.CoreV1Api, logs: TextIO | None = None +) -> list[Any] | None: + """Lists the live pods of the worker pool. + + The pool lives in one namespace by convention and the `WorkerPool` CRD is + namespaced, so this is a single scoped read. The listing is filtered + server-side by label and served from the watch cache. Use + --no-cluster-facts to skip discovery entirely. + + Returns None only when the read failed. An empty list is a reading: the + namespace holds no live worker pods. + """ + try: + pods = v1.list_namespaced_pod( + namespace=WORKER_POOL_NAMESPACE, + label_selector=WORKER_POOL_LABEL, + resource_version="0", + _request_timeout=API_TIMEOUT_SECONDS, + ).items + return [p for p in pods if p.status.phase in LIVE_POD_PHASES] + except Exception as e: + # An ApiException prints its whole HTTP response, so log the reason on + # its own. Anything without one logs itself. + reason = getattr(e, "reason", e) + _log(logs, + f"Notice: could not list pods in {WORKER_POOL_NAMESPACE}: {reason}") + return None + + +def get_cluster_hardware_facts(logs: TextIO | None = None) -> dict[str, Any]: + """Reads the worker pool size and the capacity of the nodes it runs on. + + Capacity is scoped to the nodes carrying worker pods, so a cluster that + keeps its infrastructure on a separate pool does not count that pool's + cores and memory against the density frontiers. + + Never raises: a trial must still publish its results when the cluster is + unreadable, so any failure leaves the affected facts as None. + """ + facts: dict[str, Any] = dict(EMPTY_FACTS) + if not _load_kube_config(logs): + return facts + + v1 = client.CoreV1Api() + + pods = _list_worker_pods(v1, logs) + + if pods is None: + # Without a pod set there is no node set, so capacity stays unmeasured + # rather than falling back to every node in the cluster. + return facts + + facts["worker_pod_count"] = len(pods) + + try: + # A Pending pod may not be scheduled yet, so it counts toward the pool + # size without contributing a node. + worker_nodes = {p.spec.node_name for p in pods if p.spec.node_name} + # resource_version="0" is served from the apiserver's watch cache + # rather than etcd, avoiding a quorum read on large clusters. + nodes = v1.list_node( + resource_version="0", _request_timeout=API_TIMEOUT_SECONDS + ).items + node_count = 0 + total_cores = 0.0 + total_ram_bytes = 0 + machine_types = set() + for node in nodes: + metadata = node.metadata + if metadata is None or metadata.name not in worker_nodes: + continue + node_count += 1 + allocatable = node.status.allocatable or {} + total_cores += float(parse_quantity(allocatable["cpu"])) + total_ram_bytes += int(parse_quantity(allocatable["memory"])) + machine_type = (metadata.labels or {}).get(MACHINE_TYPE_LABEL) + if machine_type: + machine_types.add(machine_type) + facts["node_count"] = node_count + facts["allocatable_cores"] = round(total_cores, 2) + # GiB, as the apiserver and kubectl quote it. + facts["allocatable_ram_gb"] = round(total_ram_bytes / (1024**3), 2) + # Kept so results stay comparable across hardware changes. A mixed pool + # is a sorted comma-joined list rather than one node picked at random. + facts["machine_type"] = ",".join(sorted(machine_types)) or None + except Exception as e: + reason = getattr(e, "reason", e) + _log(logs, f"Notice: could not read node capacity: {reason}") + + return facts + + +def append_trial_summary( + jsonl_path: Path, + stats_csv: Path, + stats_history_csv: Path, + args: argparse.Namespace, + data_ts: str, + facts: dict[str, Any], + logs: TextIO | None = None, +) -> None: + # Locust's own User Count samples. The -u flag is a request; under a custom + # load shape what actually ran is whatever the shape asked for. + observed: list[float] = [] + if stats_history_csv.exists(): + try: + with open(stats_history_csv, encoding="utf-8") as f: + for row in csv.DictReader(f): + if row.get("Name", "") not in ("", "Aggregated", "Total"): + continue + try: + u = float(row.get("User Count", "")) + except (TypeError, ValueError): + continue + if u > 0: + observed.append(u) + except Exception as e: + _log(logs, f"Notice: could not read user counts: {e}") + # A read that threw partway leaves a truncated sample behind, and + # a truncated sample understates the peak without looking wrong. + observed = [] + + # The flag stands in only when no sample was read at all. + peak_users = max(observed) if observed else args.users + + node_count = facts.get("node_count") + cores = facts.get("allocatable_cores") + ram_gb = facts.get("allocatable_ram_gb") + pod_count = facts.get("worker_pod_count") + + actors_per_node = round(peak_users / node_count, 2) if node_count else None + actors_per_vcpu = round(peak_users / cores, 2) if cores else None + actors_per_gb_ram = round(peak_users / ram_gb, 2) if ram_gb else None + + # Actors per pod across every sample, ramp-up included. Under a load shape + # there is no one target to measure steadiness against, so the + # distribution covers the whole run. + actors_per_pod_p50, actors_per_pod_p90, actors_per_pod_p99 = None, None, None + if observed and pod_count: + ratios = sorted(round(u / pod_count, 4) for u in observed) + n = len(ratios) + actors_per_pod_p50 = round(ratios[int(n * 0.50)], 2) + actors_per_pod_p90 = round(ratios[min(int(n * 0.90), n - 1)], 2) + actors_per_pod_p99 = round(ratios[min(int(n * 0.99), n - 1)], 2) + + # One ratio per row Locust reported, so a test's own operation names carry + # through. Absent when the test has no such row, null when it ran nothing. + failure_ratios: dict[str, float | None] = {"aggregate_failure_ratio": None} + if stats_csv.exists(): + try: + with open(stats_csv, encoding="utf-8") as f: + for row in csv.DictReader(f): + name = row.get("Name", "") + reqs = row.get("Request Count") + fails = row.get("Failure Count") + # Both columns required so a missing one is not read as zero. + if not name or reqs is None or fails is None: + continue + requests, failures = int(reqs), int(fails) + ratio = round(failures / requests, 4) if requests else None + if name == "Aggregated": + failure_ratios["aggregate_failure_ratio"] = ratio + continue + key = re.sub(r"([a-z0-9])([A-Z])", r"\1_\2", name) + key = re.sub(r"([A-Z]+)([A-Z][a-z])", r"\1_\2", key) + key = re.sub(r"[^a-z0-9]+", "_", key.lower()).strip("_") + failure_ratios[f"{key}_failure_ratio"] = ratio + except Exception as e: + _log(logs, f"Notice: could not parse {stats_csv}: {e}") + failure_ratios = {"aggregate_failure_ratio": None} + else: + _log(logs, f"Notice: {stats_csv} not found; failure ratios unknown") + + summary_entry = { + "timestamp": data_ts, + "tag": args.tag, + "test_name": args.name, + "metric": "trial_summary", + "measurements": { + **{k: facts.get(k) for k in EMPTY_FACTS}, + "actors_per_node": actors_per_node, + "actors_per_vcpu": actors_per_vcpu, + "actors_per_gb_ram": actors_per_gb_ram, + "actors_per_pod_p50": actors_per_pod_p50, + "actors_per_pod_p90": actors_per_pod_p90, + "actors_per_pod_p99": actors_per_pod_p99, + **failure_ratios, + }, + } + with open(jsonl_path, "a", encoding="utf-8") as f: + f.write(json.dumps(summary_entry) + "\n") + _log(logs, f"Appended trial_summary to {jsonl_path}") diff --git a/benchmarking/locust/manifests/locust.yaml b/benchmarking/locust/manifests/locust.yaml index 36374e61b4..d2cf6b3c51 100644 --- a/benchmarking/locust/manifests/locust.yaml +++ b/benchmarking/locust/manifests/locust.yaml @@ -202,3 +202,50 @@ roleRef: kind: Role name: atelet-endpointslices apiGroup: rbac.authorization.k8s.io +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRole +metadata: + name: locust-hardware-discovery +rules: +- apiGroups: [""] + resources: ["nodes"] + # list only: the runner reads whole collections, never a single object. + verbs: ["list"] +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRoleBinding +metadata: + name: locust-hardware-discovery +roleRef: + apiGroup: rbac.authorization.k8s.io + kind: ClusterRole + name: locust-hardware-discovery +subjects: +- kind: ServiceAccount + name: default + namespace: benchmarking +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: Role +metadata: + name: locust-worker-pods + namespace: benchmark-workloads +rules: +- apiGroups: [""] + resources: ["pods"] + verbs: ["list"] +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: RoleBinding +metadata: + name: locust-worker-pods + namespace: benchmark-workloads +roleRef: + apiGroup: rbac.authorization.k8s.io + kind: Role + name: locust-worker-pods +subjects: +- kind: ServiceAccount + name: default + namespace: benchmarking diff --git a/benchmarking/locust/requirements.txt b/benchmarking/locust/requirements.txt index 033e5a7422..dc7449831c 100644 --- a/benchmarking/locust/requirements.txt +++ b/benchmarking/locust/requirements.txt @@ -24,3 +24,4 @@ opentelemetry-exporter-otlp opentelemetry-instrumentation-grpc opentelemetry-instrumentation-requests google-cloud-storage +kubernetes diff --git a/benchmarking/locust/runner.py b/benchmarking/locust/runner.py index 1996e91d0d..923d88d24f 100644 --- a/benchmarking/locust/runner.py +++ b/benchmarking/locust/runner.py @@ -41,8 +41,13 @@ import time from datetime import datetime, timezone from pathlib import Path -from typing import IO, TextIO +from typing import IO, Any, TextIO +from cluster_facts import ( + EMPTY_FACTS, + append_trial_summary, + get_cluster_hardware_facts, +) from common.boomer_config import build_config_json # Path inside the locust image to the boomer-worker binary baked in by @@ -108,6 +113,16 @@ def parse_args() -> argparse.Namespace: "default of 1." ), ) + p.add_argument( + "--cluster-facts", + action=argparse.BooleanOptionalAction, + default=True, + help=( + "Read node capacity and worker pod count from the Kubernetes API " + "after the run to derive density frontiers. Pass " + "--no-cluster-facts to skip Kubernetes API discovery" + ), + ) args, extra = p.parse_known_args() args.locust_extra = extra return args @@ -406,6 +421,16 @@ def upload(src: Path, dest: str) -> None: shutil.copy(src, dest_path) +def collect_cluster_facts( + args: argparse.Namespace, logs: TextIO +) -> dict[str, Any]: + """Returns cluster hardware facts, or empty facts when discovery is off.""" + if not args.cluster_facts: + tee(logs, "Skipping cluster hardware discovery (--no-cluster-facts)") + return dict(EMPTY_FACTS) + return get_cluster_hardware_facts(logs) + + def main() -> None: args = parse_args() now = datetime.now(timezone.utc) @@ -460,6 +485,33 @@ def main() -> None: else: tee(logs, f"Stats CSV {stats_csv} not produced; skipping JSONL") + # The density frontiers are additive. They are kept out of the block + # above so that a failure here cannot discard the measurements the + # trial actually came for. + stats_history_csv = work_dir / f"{args.name}_stats_history.csv" + # Seeded up front so that a discovery failure still leaves a usable + # value for the summary below. + facts = dict(EMPTY_FACTS) + try: + facts = collect_cluster_facts(args, logs) + except Exception as e: + tee(logs, f"Warning: Failed to read cluster facts: {e}") + + # The frontiers divide by user counts, so they need the CSV. + if stats_generated: + try: + append_trial_summary( + jsonl_path, + stats_csv, + stats_history_csv, + args, + data_ts, + facts, + logs, + ) + except Exception as e: + tee(logs, f"Warning: Failed to record cluster facts: {e}") + status_path.write_text( json.dumps( {"locust_exit_code": exit_code, "stats_generated": stats_generated} diff --git a/benchmarking/locust/unit_tests/test_cluster_facts.py b/benchmarking/locust/unit_tests/test_cluster_facts.py new file mode 100644 index 0000000000..1e40354e33 --- /dev/null +++ b/benchmarking/locust/unit_tests/test_cluster_facts.py @@ -0,0 +1,280 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Unit tests for cluster_facts.py. + +Run via: python3 benchmarking/locust/unit_tests/test_cluster_facts.py +Never contacts a cluster. Nodes and pods are stand-in objects handed to a +mocked CoreV1Api. Needs the kubernetes client: +pip install -r benchmarking/locust/requirements.txt +""" + +import argparse +import contextlib +import io +import json +import sys +import tempfile +import unittest +from pathlib import Path +from types import SimpleNamespace +from unittest import mock + +from kubernetes.client.rest import ApiException + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import cluster_facts +import runner + +# Real apiserver quantity strings: 3920m -> 3.92 cores, 13591700Ki -> 12.96 GiB. +NODE_CPU, NODE_MEMORY = "3920m", "13591700Ki" + +# One successful discovery. 10 users against 5 worker pods puts actors per pod at 2.0. +FACTS = {"machine_type": "c3-standard-4", "node_count": 1, + "allocatable_cores": 3.92, "allocatable_ram_gb": 12.96, + "worker_pod_count": 5} + +STATS_HEADER = "Type,Name,Request Count,Failure Count\n" +ARGV = ["runner.py", "-f", "tests/glutton.py", "-t", "1m", "-u", "10", + "--tag", "unit", "--name", "unit-run", "--dest", "/tmp/unit"] + + +def node(machine_type="c3-standard-4", name="node-a"): + labels = {} if machine_type is None else { + cluster_facts.MACHINE_TYPE_LABEL: machine_type} + return SimpleNamespace( + metadata=SimpleNamespace(labels=labels, name=name), + status=SimpleNamespace(allocatable={"cpu": NODE_CPU, "memory": NODE_MEMORY})) + + +def pod(phase="Running", node_name="node-a"): + return SimpleNamespace(spec=SimpleNamespace(node_name=node_name), + status=SimpleNamespace(phase=phase)) + + +def fake_api(nodes=None, ns_pods=None): + """Stand-in CoreV1Api; None raises 403 Forbidden.""" + def forbidden(*_args, **_kwargs): + raise ApiException(status=403, reason="Forbidden") + + api = mock.Mock() + for attr, items in (("list_node", nodes), ("list_namespaced_pod", ns_pods)): + if items is None: + getattr(api, attr).side_effect = forbidden + else: + getattr(api, attr).return_value = SimpleNamespace(items=items) + return api + + +def discover(api): + with mock.patch.object(cluster_facts, "_load_kube_config", return_value=True), \ + mock.patch.object(cluster_facts.client, "CoreV1Api", return_value=api), \ + contextlib.redirect_stdout(io.StringIO()): + return cluster_facts.get_cluster_hardware_facts() + + +def summarize(facts, directory, stats=STATS_HEADER + ",Aggregated,100,25\n", + users=10, user_counts=None): + """Writes CSV inputs and returns the emitted trial_summary row.""" + d = Path(directory) + if stats is not None: + (d / "stats.csv").write_text(stats) + if user_counts is None: + user_counts = [users] * 61 + (d / "stats_history.csv").write_text( + "Timestamp,User Count,Type,Name,Requests/s,Failures/s\n" + + "".join(f"{1788914584 + i},{u},,Aggregated,1.0,0.0\n" + for i, u in enumerate(user_counts))) + out = d / "out.jsonl" + with contextlib.redirect_stdout(io.StringIO()): + cluster_facts.append_trial_summary( + out, d / "stats.csv", d / "stats_history.csv", + argparse.Namespace(users=users, tag="unit", name="unit-run"), + "2026-01-01", facts) + return json.loads(out.read_text().splitlines()[0]) + + +def parse(*extra): + with mock.patch.object(sys, "argv", ARGV + list(extra)): + return runner.parse_args() + + +class ClusterFactsTest(unittest.TestCase): + def test_node_capacity(self): + facts = discover(fake_api( + nodes=[node(name="node-a"), node(None, name="node-b"), + node("n2-standard-8", name="node-c")], + ns_pods=[pod(node_name="node-a"), pod(node_name="node-b"), + pod(node_name="node-c")])) + self.assertEqual(facts["node_count"], 3) + self.assertEqual(facts["allocatable_cores"], 11.76) # 3 x 3.92 + self.assertEqual(facts["allocatable_ram_gb"], 38.89) + self.assertEqual(facts["machine_type"], "c3-standard-4,n2-standard-8") + + def test_worker_pod_count(self): + pods = [pod("Running"), pod("Pending"), pod("Succeeded"), pod("Failed")] + api = fake_api(nodes=[node()], ns_pods=pods) + self.assertEqual(discover(api)["worker_pod_count"], 2) + api.list_namespaced_pod.assert_called_once_with( + namespace="benchmark-workloads", + label_selector="ate.dev/worker-pool", + resource_version="0", + _request_timeout=5, + ) + api.list_node.assert_called_once_with( + resource_version="0", + _request_timeout=5, + ) + + def test_capacity_is_scoped_to_nodes_running_worker_pods(self): + # Excludes nodes without worker pods (e.g. infra-a). + facts = discover(fake_api( + nodes=[node(name="node-a"), node(name="node-b"), + node(name="infra-a")], + ns_pods=[pod(node_name="node-a"), pod(node_name="node-b"), + pod(node_name="node-b")])) + self.assertEqual(facts["worker_pod_count"], 3) + self.assertEqual(facts["node_count"], 2) + self.assertEqual(facts["allocatable_cores"], 7.84) # 2 x 3.92 + + def test_zero_worker_pods_is_a_reading(self): + # Empty pool -> 0; denied read -> None. + facts = discover(fake_api(nodes=[node()], ns_pods=[])) + self.assertEqual(facts["worker_pod_count"], 0) + self.assertEqual(facts["node_count"], 0) + + denied = discover(fake_api(nodes=[node()], ns_pods=None)) + self.assertIsNone(denied["worker_pod_count"]) + self.assertIsNone(denied["node_count"]) + + def test_unreadable_facts_are_none(self): + # Nodes denied -> capacity None, pod count still recorded. + facts = discover(fake_api(nodes=None, ns_pods=[pod(), pod()])) + self.assertIsNone(facts["node_count"]) + self.assertIsNone(facts["allocatable_cores"]) + self.assertEqual(facts["worker_pod_count"], 2) + + # Everything denied or no credentials -> EMPTY_FACTS. + self.assertEqual(discover(fake_api()), cluster_facts.EMPTY_FACTS) + with mock.patch.object(cluster_facts, "_load_kube_config", return_value=False): + self.assertEqual(cluster_facts.get_cluster_hardware_facts(), + cluster_facts.EMPTY_FACTS) + + def test_flags(self): + extra = parse("--no-cluster-facts", "--max-wait-time", "1.0") + self.assertNotIn("--no-cluster-facts", extra.locust_extra) + self.assertEqual(extra.locust_extra, ["--max-wait-time", "1.0"]) + + def test_no_cluster_facts_skips_the_api(self): + def tripwire(*_args, **_kwargs): + raise AssertionError("Kubernetes was contacted with --no-cluster-facts") + + with mock.patch.object(cluster_facts.client, "CoreV1Api", tripwire), \ + mock.patch.object( + cluster_facts.config, "load_incluster_config", tripwire + ), \ + mock.patch.object(cluster_facts.config, "load_kube_config", tripwire), \ + contextlib.redirect_stdout(io.StringIO()): + facts = runner.collect_cluster_facts(parse("--no-cluster-facts"), + io.StringIO()) + self.assertEqual(facts, cluster_facts.EMPTY_FACTS) + + with mock.patch.object(runner, "get_cluster_hardware_facts", + return_value={"node_count": 1}) as discovery, \ + contextlib.redirect_stdout(io.StringIO()): + runner.collect_cluster_facts(parse(), io.StringIO()) + discovery.assert_called_once() + + def test_trial_summary(self): + with tempfile.TemporaryDirectory() as td: + row = summarize(FACTS, td) + self.assertEqual(row["metric"], "trial_summary") + self.assertEqual(set(row), {"timestamp", "tag", "test_name", "metric", + "measurements"}) + m = row["measurements"] + self.assertEqual({k: m[k] for k in FACTS}, FACTS) + self.assertEqual(m["actors_per_node"], 10.0) # 10 users / 1 node + self.assertEqual(m["actors_per_vcpu"], 2.55) # 10 / 3.92 + self.assertEqual(m["actors_per_gb_ram"], 0.77) # 10 / 12.96 + + # Unmeasured facts keep the same keys with None values. + with tempfile.TemporaryDirectory() as td: + row = summarize(dict(cluster_facts.EMPTY_FACTS), td) + self.assertLessEqual(set(cluster_facts.EMPTY_FACTS), + set(row["measurements"])) + for key in ("actors_per_node", "actors_per_vcpu", "actors_per_gb_ram", + "actors_per_pod_p50", "actors_per_pod_p90", + "actors_per_pod_p99"): + self.assertIsNone(row["measurements"][key]) + + def test_actors_per_pod_percentiles(self): + with tempfile.TemporaryDirectory() as td: + row = summarize(FACTS, td, users=100, + user_counts=[1, 50, 89] + list(range(100, 300))) + m = row["measurements"] + self.assertEqual([m["actors_per_pod_p50"], m["actors_per_pod_p90"], + m["actors_per_pod_p99"]], + [39.6, 55.8, 59.4]) # users 198, 279, 297 over 5 pods + + # Uses observed peak (60) rather than requested -u (10). + with tempfile.TemporaryDirectory() as td: + row = summarize(FACTS, td, users=10, user_counts=[20, 40, 60]) + self.assertEqual(row["measurements"]["actors_per_node"], 60.0) + self.assertEqual(row["measurements"]["actors_per_pod_p50"], 8.0) + + # Empty history falls back to -u for frontiers and None for percentiles. + with tempfile.TemporaryDirectory() as td: + row = summarize(FACTS, td, users=10, user_counts=[]) + for key in ("actors_per_pod_p50", "actors_per_pod_p90", + "actors_per_pod_p99"): + self.assertIsNone(row["measurements"][key]) + self.assertEqual(row["measurements"]["actors_per_node"], 10.0) + + def test_failure_ratio(self): + def ratio(stats): + with tempfile.TemporaryDirectory() as td: + return summarize(FACTS, td, stats)["measurements"][ + "aggregate_failure_ratio"] + + self.assertEqual(ratio(STATS_HEADER + ",Aggregated,100,25\n"), 0.25) + self.assertEqual(ratio(STATS_HEADER + ",Aggregated,1708,0\n"), 0.0) + self.assertIsNone(ratio(STATS_HEADER + ",Aggregated,100\n")) # truncated + self.assertIsNone(ratio(STATS_HEADER + ",Aggregated,bad,5\n")) # corrupt int + self.assertIsNone(ratio(STATS_HEADER + ",Aggregated,0,0\n")) # 0/0 + self.assertIsNone(ratio(None)) # file absent + + def test_per_rpc_failure_ratios(self): + # First resume is its own Locust row, so it gets its own key. + stats = (STATS_HEADER + + "grpc,ResumeActor,100,2\n" + + "grpc,ResumeActorFirstResume,100,99\n" + + "grpc,SuspendActor,200,0\n" + + ",Aggregated,400,101\n") + with tempfile.TemporaryDirectory() as td: + f = summarize(FACTS, td, stats)["measurements"] + self.assertEqual(f["resume_actor_failure_ratio"], 0.02) # 2 / 100 + self.assertEqual(f["resume_actor_first_resume_failure_ratio"], 0.99) # 99 / 100 + self.assertEqual(f["suspend_actor_failure_ratio"], 0.0) + self.assertEqual(f["aggregate_failure_ratio"], 0.2525) # 101 / 400 + + # An RPC the test never ran has no key at all. + with tempfile.TemporaryDirectory() as td: + row = summarize(FACTS, td, STATS_HEADER + ",Aggregated,10,0\n") + self.assertNotIn("resume_actor_failure_ratio", row["measurements"]) + self.assertNotIn("suspend_actor_failure_ratio", row["measurements"]) + + +if __name__ == "__main__": + unittest.main()