diff --git a/AGENTS.md b/AGENTS.md index 95c567a..3a414a7 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -84,16 +84,17 @@ Breaking (a major release): 60,000 source ports, about `cps × (hold + TIME_WAIT)`. - SIPp names its result files after its pid (`__`); the files of another load generator get their worker number in front of the pid so two machines never collide. -- Routr routes a call to the address in the top Via of the callee's REGISTER, so the callees - always run on the first load generator. +- Routr routes a call to the address in the top Via of the callee's REGISTER, so each callee + process registers its own share of the users from its own address and port, and calls reach + the process that holds the user. ### performance-engineer - SIPp is single-threaded: about 350 calls/s per process over TCP, so a bigger load generator needs more `load.instances`, not more cores per process. - A capacity step is invalid when a SIPp process uses more than 80% of a core or the placed - rate is outside 95% to 105% of the offered one. The sampler watches the first load - generator's SIPp processes only, not the workers'. + rate is outside 95% to 105% of the offered one. Every load generator runs a sampler, and + the callees count too (one per instance). ### release-engineer @@ -118,5 +119,6 @@ Breaking (a major release): - 2026-11-09: Every benchmark runs the same load generator layout so results compare: `load.instances` is the total SIPp uac processes, spread round robin over the load generators (two smaller droplets rather than one big one); benchmarks use 2, the Kubernetes - ones 8. The agent and the callees stay on the first; the others only run uac and take - commands over SSH. + ones 8. The agent stays on the first; the others only run SIPp (uac and callees) and take + commands over SSH. There are as many callee processes as instances, each with its share of + the users. diff --git a/README.md b/README.md index 7d6fa1c..ddcd22e 100644 --- a/README.md +++ b/README.md @@ -8,10 +8,10 @@ report page. perfkit benchmarks Fonoster projects (Routr first, then Roomr and Fonoster) the same way and publishes the results in one format. -- **Two droplets per run.** Each run creates two dedicated-CPU DigitalOcean droplets in the same - region and private VPC: the unit under test, and a load generator that drives it with SIPp. - The images are pinned by digest when the run starts, and both droplets are deleted on every - exit path. +- **A unit under test and its load generators.** Each run creates dedicated-CPU DigitalOcean + droplets in the same region and private VPC: the unit under test, and one to eight load + generators (`infra.loadgen.count`) that drive it with SIPp. The images are pinned by digest + when the run starts, and every droplet is deleted on every exit path. - **A Kubernetes cluster when a benchmark asks for one.** A benchmark with `uut.helm` installs a Helm chart on a DOKS cluster created for the run in the same VPC (kind with MetalLB on `--local`), and the page labels each benchmark with its target. The cluster is deleted with its @@ -27,6 +27,54 @@ perfkit/v1`). See [docs/perfkit-yaml.md](docs/perfkit-yaml.md) for every field a [docs/report-v1.md](docs/report-v1.md) for `report.json` and `runs.json`. The JSON Schema files are in [schemas/](schemas/). +## How it works + +`perfkit run` creates the machines and starts an agent on the first load generator. The agent +runs every benchmark on its own, publishes `live.json` while it runs, and keeps the raw files. +`perfkit collect` fetches them, writes `report.json` and deletes everything. + +```mermaid +flowchart LR + subgraph CI["GitHub Actions or a laptop"] + RUN["perfkit run
reads perfkit.yaml"] + COL["perfkit collect
analyze, report.json"] + end + + subgraph DO["DigitalOcean, one private VPC per run"] + direction TB + LG1["Load generator 1
agent, sampler,
SIPp callers and callees"] + LGN["Load generators 2 to 8
sampler,
SIPp callers and callees"] + UUT["Unit under test
one server (Docker Compose)
or a DOKS cluster (Helm)"] + LG1 -- "SSH, run-only key" --> LGN + LG1 -- SIP --> UUT + LGN -- SIP --> UUT + end + + RUN -- "1 create droplets, firewall,
cluster; start the agent" --> DO + LG1 -- "2 live.json" --> LIVE["Live page"] + DO -- "3 raw files" --> COL + COL -- "4 delete everything" --> DO + COL --> SITE["Report page
perfkit render"] +``` + +Every benchmark spreads `load.instances` SIPp callers and as many callees round robin over the +load generators. The callees register their share of the users, then answer the calls Routr +routes to them; no load generator talks to another during the load. + +```mermaid +flowchart TD + A["Next benchmark"] --> B["Deploy the unit under test
compose up, or helm install and seed"] + B --> C{"Every callee
registered its share?"} + C -- no --> X["Abort the benchmark"] + C -- yes --> K{kind} + K -- capacity --> D["Offer a rate for one step
(30 s), then add a step"] + D --> F{"p99, failed calls,
any load generator saturated?"} + F -- ok --> D + F -- "a limit broke twice" --> G["Limit: the last good step"] + F -- saturated --> H["Lower bound: the load generator
ran out before the unit under test"] + K -- soak --> S["Hold a share of the measured limit
for hours; memory, failed calls, restarts"] +``` + ## Prerequisites - Node 22 or later, and npm 11 to run perfkit from a git tag (see below). @@ -401,8 +449,8 @@ failed call is one SIPp counts as failed, and CPU is in % of one core, so 100 is ### Capacity SIPp adds load in steps, such as 50 more calls/s every minute. With callees (`load.register`), -the ramp starts only once every callee has registered, with the soak's timeout (`users / 20` + -60 s) and abort reason. The uac gives up on a message it expects after 10 s, and a call that +the ramp starts only once every callee process has registered its whole share, with the soak's +timeout (`users / 20` + 60 s) and abort reason. The uac gives up on a message it expects after 10 s, and a call that timed out counts as failed; with a hold, a call's teardown failure is counted when it happens, up to `hold` + 10 s after the call starts, so `every` should be at least 4 × (hold + 10 s). For each step: @@ -420,7 +468,7 @@ step: perfkit checks the steps in order: 1. **Is the measurement valid?** A step is invalid when the load generator, not the server, was - the bottleneck: any SIPp role (the uac, or the callees `sipp-register` and `sipp-uas`) used + the bottleneck: any SIPp process (a uac instance, or a callee, `sipp-register` and `sipp-uas`) used more than 80% of one core, placed less than 95% of the offered rate, placed more than 105% of it (a burst of calls it owed from earlier steps), or the load generator dropped UDP packets. Then the verdict is **inconclusive**, the reason @@ -444,7 +492,7 @@ A soak holds a fixed rate for hours to find leaks. With `rate.fromCapacity`, the (when it was inconclusive because the load generator gave out, the highest step before that, if every one passed, as a lower bound), or `fallbackCps` when it found none; the progress log and `load.rateSource` in report.json say which. -On the cluster, a restart is a container restart of a sampled pod (its restart count went up). +On the cluster, a restart is a container restart of a sampled pod (its restart count went up) or a pod replaced by another; a sample the Metrics API left out is not a restart. - **Warm-up excluded.** Nothing measured during `warmup` counts toward any rule. - **Too little data.** When the samples after the warm-up cover less than 90% of `duration`, the @@ -456,7 +504,8 @@ On the cluster, a restart is a container restart of a sampled pod (its restart c - **Callees not registered.** With a register role, the uac starts only once every callee has registered. When they have not within `users / 20` + 60 seconds, no call is placed and the verdict is **aborted** (`only 150 of 200 callees registered within 70 s, so no call was - placed`). + placed`). With several callee processes the reason names the one that fell short (`callee 2 + registered 10 of 67 within 64 s, so no call was placed`). - **Memory growth.** For each process, the least-squares slope of its memory (RSS) in MB per hour. It fails at the threshold or above: 5.0 MB/h fails a 5 MB/h limit. The headline is the steepest process. @@ -544,7 +593,7 @@ it to publish (or clear) a project's site before its first published run: ```console $ npx perfkit render --config perf/perfkit.yaml --out out/benchmarks -wrote out/benchmarks/app/0.1.12/app.js (perfkit 0.1.12) +wrote out/benchmarks/app/0.1.13/app.js (perfkit 0.1.13) wrote out/benchmarks/data/runs.json (0 runs) wrote out/benchmarks/data/site.json (live: https://perfkit.ngrok.app/live.json) wrote out/benchmarks/index.html (no runs yet) diff --git a/docs/design-loadgen-count.md b/docs/design-loadgen-count.md index e99e915..5e61e78 100644 --- a/docs/design-loadgen-count.md +++ b/docs/design-loadgen-count.md @@ -20,16 +20,24 @@ infra: ``` - `count` droplets, tagged like the load generator today; one is the primary (it runs the - agent, the samplers' collection and the callees), the others are workers. + agent and the samplers' collection), the others are workers. - Each worker runs its share of the uac: `instances` and the rate are per worker, so a benchmark offers `count × rate`. The users file is split across all `count × instances` processes. - Raw results: every worker's trace files are copied into the benchmark's raw directory (distinct names, as SIPp's pid already makes them), and `readSippResults` already merges - every pid. The sampler runs on each worker and writes `sampler-loadgen-.jsonl`; the - analysis names processes `sipp-uac@` so saturation is judged per process. -- The callees (register and uas) stay on the primary: Routr routes a call to the address of the - REGISTER's top Via. + every pid. The sampler runs on each worker and writes `sampler-worker-.jsonl` and + `udp-worker-.jsonl`, copied back with the rest; process names are one per instance + (`sipp-uac-3`, `sipp-register-2`), so saturation is judged per process on any machine, + and UDP drops on any machine invalidate the step. The report's `run.infra.loadgen.count` + says how many droplets there were. +- The callees (register and uas) are split like the uac: `instances` callee processes, each + registering its share of the users (round robin) from its own port and address, spread over + the load generators. Routr routes a call to the address of the REGISTER's top Via, so it goes + to the callee that registered the user. Before the ramp every callee must have registered its + share, or the benchmark aborts (`callee 2 registered 10 of 67 ...`). One callee process tops + out near 2,900 calls/s over UDP (about 17,000 messages/s), which capped RoutrOne's first + full run. ## What it touches diff --git a/docs/perfkit-yaml.md b/docs/perfkit-yaml.md index 6c17392..3e5f6dd 100644 --- a/docs/perfkit-yaml.md +++ b/docs/perfkit-yaml.md @@ -45,7 +45,7 @@ Rules for the whole file: | `infra.region` | string | yes | | A DigitalOcean region slug, like `nyc3`. Both droplets run there. | | `infra.uut.size` | string | yes | | Droplet size slug of the unit under test, a dedicated-CPU size like `c-8`. | | `infra.loadgen.size` | string | yes | | Droplet size slug of the load generator, like `c-16`. | -| `infra.loadgen.count` | integer | no | `1` | Load generator droplets, 1 to 8, all of `size`. The first runs the agent and the callees; the others only run uac SIPp processes. | +| `infra.loadgen.count` | integer | no | `1` | Load generator droplets, 1 to 8, all of `size`. The first runs the agent; the uac and callee SIPp processes are spread over all of them. | | `infra.cluster` | object | yes, when a benchmark has `uut.helm` | none | The Kubernetes cluster of the helm benchmarks. A run creates it only when a selected benchmark has `uut.helm`. | | `infra.cluster.size` | string | yes, if `cluster` is set | | Droplet size slug of the DOKS node pool, like `c-8`. | | `infra.cluster.nodes` | integer, 1 to 20 | yes, if `cluster` is set | | Nodes in the pool. With `--local`, the kind cluster's nodes: a control plane, then workers. | @@ -152,7 +152,7 @@ the other kind rejects them. | `load.register.users` | path | yes, if `register` is set | | The agents to register (CSV). | | `load.register.expires` | integer (seconds) | no | `300` | The registration lifetime. | | `load.uas.scenario` | path | yes, if `uas` is set | | `load.uas` is optional, for either kind: the SIPp scenario that answers calls. | -| `load.instances` | integer, 1 to 64 | no | `1` | How many SIPp processes place the calls. SIPp runs on one core, so a single process tops out at about 350 calls/s over TCP and a few thousand over UDP, however big the load generator is. With `instances: N`, perfkit starts N uac processes on the load generator, each with 1/N of the rate (`-r`, `-rate_increase`, `-rate_max`) and of the open calls (`-l`), its own local port (5080, 5081, ...) and Call-ID prefix (`uac-`, `uac2-`, ...), and its own share of `uac.users` (every N-th row, so a user belongs to one process). The callees (`register`, `uas`) stay one process. The report sums all of them. A capacity step is invalid when any one instance is over 80% of its core; set `instances` to about the cores you want to use, leaving a few for the callees and the sampler. A local cluster (`--local`, fixed SIPp addresses) accepts only `1`. | +| `load.instances` | integer, 1 to 64 | no | `1` | How many SIPp processes place the calls. SIPp runs on one core, so a single process tops out at about 350 calls/s over TCP and a few thousand over UDP, however big the load generator is. With `instances: N`, perfkit starts N uac processes on the load generator, each with its share of the rate (`-r`, `-rate_increase`, `-rate_max`) in whole numbers that add up to the rate (50 over 8 is six 6s and two 7s, as SIPp's `-rate_increase` takes only integers; `rate.start` must be at least `instances`) and of the open calls (`-l`), its own local port (5080, 5081, ...) and Call-ID prefix (`uac-`, `uac2-`, ...), and its own share of `uac.users` (every N-th row, so a user belongs to one process). With a `register`, the callees are split the same way: `instances` callee processes (`sipp-register`, `sipp-register-2`, ...), each with its own share of `register.users` (every N-th row, so a callee belongs to one process), its own port (5070, then 5200 and up), Call-ID prefix and address; Routr sends a call to the process the callee registered from. A `uas` alone, with no `register`, stays one process. The report sums all of them. A capacity step is invalid when any one instance is over 80% of its core; set `instances` to about the cores you want to use, leaving a few for the callees and the sampler. A local cluster (`--local`, fixed SIPp addresses) accepts only `1`. | | `load.hold` | duration | no | `0s` | How long each call stays up before BYE. | | `load.unit` | string | no | `calls` | What one SIPp call is, as a lowercase plural noun shown on the page (`calls`, `messages`). Copied into each report.json benchmark as `unit`. | | `load.rate.start` | number (calls/s) | *capacity*: yes | | The first step's rate. | @@ -322,8 +322,9 @@ connections per load generator address, which is `cps × (hold + TIME_WAIT)`. `infra.loadgen.count` adds load generator droplets, each with its own address and its own pool of source ports. `load.instances` stays the total number of uac SIPp processes and the rate stays the benchmark's: the instances are spread round robin over the droplets (the -first on the first), each with its share of the callers. Two c-8 droplets cost about what one -c-16 does and have twice the ports. The callees stay on the first droplet, and the others' -SIPp files are copied back to it every 30 s and when the benchmark ends. Limits: the sampler -watches the first droplet's SIPp processes only, so a saturated worker is not flagged. +first on the first), each with its share of the callers, and so are the callees, each with its +share of the users. Two c-8 droplets cost about what one c-16 does and have twice the ports. The +others' SIPp files and sampler files are copied back to the first every 30 s and when the +benchmark ends. Every droplet runs its own sampler, so a SIPp process over 80% of its core, or UDP +drops, on any of them makes the step invalid. The design is in `docs/design-loadgen-count.md`. diff --git a/docs/report-v1.md b/docs/report-v1.md index 5567176..993ac6f 100644 --- a/docs/report-v1.md +++ b/docs/report-v1.md @@ -54,7 +54,7 @@ Rules for both files: | `run.startedAt`, `run.endedAt` | timestamp | | | `run.infra.provider` | `digitalocean` or `local` | `local` for `perfkit run --local`, where both roles run on one machine's Docker; sizes then read `local`. | | `run.infra.region` | string | | -| `run.infra.uut`, `run.infra.loadgen` | object | `size` (slug), `vcpus` and `memoryMB` of each droplet. | +| `run.infra.uut`, `run.infra.loadgen` | object | `size` (slug), `vcpus` and `memoryMB` of each droplet. `loadgen` also has `count`, the number of load generator droplets, when there are 2 or more. | | `run.infra.os` | string | The droplets' OS, like `Ubuntu 24.04`. | | `run.infra.docker` | string | The droplets' Docker version. | | `run.infra.cluster` | object | Optional: only when the run had a Kubernetes cluster. `provider` (`digitalocean` or `kind`), `nodes`, `size` (the node slug, `local` for kind) and `version` (like `1.33.1-do.0`). | diff --git a/package-lock.json b/package-lock.json index 4dfc36c..7195fab 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "perfkit", - "version": "0.1.12", + "version": "0.1.13", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "perfkit", - "version": "0.1.12", + "version": "0.1.13", "license": "MIT", "dependencies": { "@ngrok/ngrok": "1.7.0", diff --git a/package.json b/package.json index 33970ce..7488b6f 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "perfkit", - "version": "0.1.12", + "version": "0.1.13", "description": "Benchmark harness for Fonoster projects: DigitalOcean runs, one report.json schema, one branded report page", "license": "MIT", "author": "Pedro Sanders ", diff --git a/schemas/report-v1.schema.json b/schemas/report-v1.schema.json index 4bebb09..954f4ac 100644 --- a/schemas/report-v1.schema.json +++ b/schemas/report-v1.schema.json @@ -145,6 +145,11 @@ "type": "integer", "exclusiveMinimum": 0, "maximum": 9007199254740991 + }, + "count": { + "type": "integer", + "minimum": 2, + "maximum": 9007199254740991 } }, "required": [ diff --git a/src/analyze/countRestarts.ts b/src/analyze/countRestarts.ts index b6bbd7c..4450db3 100644 --- a/src/analyze/countRestarts.ts +++ b/src/analyze/countRestarts.ts @@ -12,12 +12,36 @@ import type { ProcessSample } from "./types.js"; * that is a pid change. A tick where the process is missing counts 1; a run * of missing ticks counts once, and the pids it comes back with are not * counted again. + * + * On the cluster (`fromCounters`) a name's `pid` is 1 plus the restarts its + * pods' counters show (see `toKubeSamplerLines`), so the restarts are the + * rises of the pid, and a tick with no sample (the metrics API had no usage + * yet) is not a restart: only a counter says a pod restarted. */ export function countRestarts( samples: readonly ProcessSample[], - processNames: readonly string[] + processNames: readonly string[], + options: { fromCounters?: boolean } = {} ): number { const uut = samples.filter((sample) => sample.host === "uut"); + if (options.fromCounters) { + return processNames.reduce((total, name) => { + const pids = uut + .filter((sample) => sample.name === name) + .sort((a, b) => a.tMs - b.tMs) + .map((sample) => sample.pid); + return ( + total + + pids.reduce( + (sum, pid, index) => + index === 0 + ? sum + : sum + Math.max(0, pid - (pids[index - 1] ?? pid)), + 0 + ) + ); + }, 0); + } const ticks = [...new Set(uut.map((sample) => sample.tMs))].sort( (a, b) => a - b ); diff --git a/src/analyze/createAnalyzeCapacity.ts b/src/analyze/createAnalyzeCapacity.ts index 1514eea..5169f65 100644 --- a/src/analyze/createAnalyzeCapacity.ts +++ b/src/analyze/createAnalyzeCapacity.ts @@ -3,6 +3,7 @@ */ import { withErrorHandlingAndValidation } from "../errors/index.js"; import { + calleeInstancesOf, type CapacityBenchmarkResult, type CapacityRule, type StoppedBy, @@ -17,6 +18,7 @@ import { listRawDir } from "./listRawDir.js"; import { type MeasuredStep, measureCapacityStep, + calleeProcessNames, uacProcessNames } from "./measureCapacityStep.js"; import { parseJsonDocument } from "./parseJsonDocument.js"; @@ -89,6 +91,7 @@ export function createAnalyzeCapacity(deps: AnalyzeDeps) { udp, uutProcesses, uacProcesses: uacProcessNames(benchmark.load.instances), + calleeProcesses: calleeProcessNames(calleeInstancesOf(benchmark.load)), unit }) ); diff --git a/src/analyze/createAnalyzeSoak.ts b/src/analyze/createAnalyzeSoak.ts index 00075ae..a4bd7f5 100644 --- a/src/analyze/createAnalyzeSoak.ts +++ b/src/analyze/createAnalyzeSoak.ts @@ -4,6 +4,7 @@ import { basename } from "node:path"; import { withErrorHandlingAndValidation } from "../errors/index.js"; import { + isHelmUut, type SoakBenchmarkResult, type SoakRule, uutProcessNames @@ -147,7 +148,9 @@ export function createAnalyzeSoak(deps: AnalyzeDeps) { const failedPct = roundTo( callSums.attempted > 0 ? (callSums.failed / callSums.attempted) * 100 : 0 ); - const restarts = countRestarts(after, uutProcesses); + const restarts = countRestarts(after, uutProcesses, { + fromCounters: isHelmUut(benchmark.uut) + }); const { thresholds } = benchmark; const growth = uutProcesses.map((name) => ({ diff --git a/src/analyze/index.ts b/src/analyze/index.ts index 6f9cf9a..5f746c2 100644 --- a/src/analyze/index.ts +++ b/src/analyze/index.ts @@ -30,6 +30,8 @@ export { durationToSeconds } from "./durationToSeconds.js"; export { leastSquaresSlope } from "./leastSquaresSlope.js"; export { memoryGrowthMBPerHour } from "./memoryGrowthMBPerHour.js"; export { + calleeProcessName, + calleeProcessNames, MAX_SIPP_CPU_PCT, measureCapacityStep, MIN_ACHIEVED_RATIO, diff --git a/src/analyze/measureCapacityStep.ts b/src/analyze/measureCapacityStep.ts index 2ce474d..d186d20 100644 --- a/src/analyze/measureCapacityStep.ts +++ b/src/analyze/measureCapacityStep.ts @@ -41,9 +41,30 @@ export function uacProcessNames(instances: number): string[] { uacProcessName(index + 1) ); } -/** The callee SIPp processes, checked against the same 80%. */ +/** The callee SIPp processes of one instance, checked against the same 80%. */ export const SIPP_CALLEE_PROCESSES = ["sipp-register", "sipp-uas"] as const; +/** + * The sampler's name for one callee instance of a role (`register` or + * `uas`): `sipp-register` for the first, `sipp-register-2` and on for the + * others. + */ +export function calleeProcessName( + role: "register" | "uas", + instance: number +): string { + return instance > 1 ? `sipp-${role}-${instance}` : `sipp-${role}`; +} + +/** The sampler names of the callee processes of `instances` callee instances. */ +export function calleeProcessNames(instances: number): string[] { + return Array.from({ length: Math.max(1, instances) }, (_, index) => + (["register", "uas"] as const).map((role) => + calleeProcessName(role, index + 1) + ) + ).flat(); +} + export interface StepInput { offeredCps: number; window: TimeWindow; @@ -53,6 +74,8 @@ export interface StepInput { uutProcesses: readonly string[]; /** The uac SIPp processes' sampler names; just `sipp-uac` by default. */ uacProcesses?: readonly string[]; + /** The callee SIPp processes' sampler names; `sipp-register` and `sipp-uas` by default. */ + calleeProcesses?: readonly string[]; unit: string; } @@ -86,7 +109,7 @@ export function measureCapacityStep(input: StepInput): MeasuredStep { const instanceCpu = (input.uacProcesses ?? [SIPP_UAC_PROCESS]).map( (name) => ({ name, - cpuPct: cpuPctOver(processSamples(input.samples, "loadgen", name), window) + cpuPct: cpuPctOver(loadgenSamples(input.samples, name), window) }) ); const busiest = instanceCpu.reduce((top, one) => @@ -102,11 +125,18 @@ export function measureCapacityStep(input: StepInput): MeasuredStep { 0 ) ); - const callee = SIPP_CALLEE_PROCESSES.map((name) => ({ - name, - cpuPct: cpuPctOver(processSamples(input.samples, "loadgen", name), window) - })).find(({ cpuPct }) => cpuPct > MAX_SIPP_CPU_PCT); - const loadgenUdpDrops = udpDropsOver(input.udp, "loadgen", window); + const callee = (input.calleeProcesses ?? SIPP_CALLEE_PROCESSES) + .map((name) => ({ + name, + cpuPct: cpuPctOver(loadgenSamples(input.samples, name), window) + })) + .find(({ cpuPct }) => cpuPct > MAX_SIPP_CPU_PCT); + // Every load generator counts: a worker that drops packets spoils the step too + const loadgenUdpDrops = [ + ...new Set( + input.udp.map(({ host }) => host).filter((host) => host !== "uut") + ) + ].reduce((sum, host) => sum + udpDropsOver(input.udp, host, window), 0); const uutUdpDrops = udpDropsOver(input.udp, "uut", window); const rate = `${formatNumber(offeredCps)} ${input.unit}/s`; const invalidReason = @@ -143,6 +173,20 @@ export function measureCapacityStep(input: StepInput): MeasuredStep { }; } +/** + * A load generator process's samples, from whichever load generator runs it: + * the process names (`sipp-uac-3`, `sipp-register-2`) are one per instance, + * so no two machines share one. + */ +function loadgenSamples( + samples: readonly ProcessSample[], + name: string +): ProcessSample[] { + return samples.filter( + (sample) => sample.host !== "uut" && sample.name === name + ); +} + function processSamples( samples: readonly ProcessSample[], host: ProcessSample["host"], diff --git a/src/analyze/rawResults.schema.ts b/src/analyze/rawResults.schema.ts index d19f296..61bde7b 100644 --- a/src/analyze/rawResults.schema.ts +++ b/src/analyze/rawResults.schema.ts @@ -58,9 +58,15 @@ export const soakMetaSchema = z.strictObject({ registerUsers: count.default(0) }); +/** The host a sampler line comes from: the UUT, a load generator or a worker. */ +const samplerHostSchema = z.union([ + z.enum(["uut", "loadgen"]), + z.templateLiteral(["worker-", z.number().int().positive()]) +]); + export const samplerLineSchema = z.strictObject({ t: timestampSchema, - host: z.enum(["uut", "loadgen"]), + host: samplerHostSchema, name: nonEmptyString, pid: positiveNumber.int({ error: "expected a whole number" }), rssKB: nonNegativeNumber, @@ -71,7 +77,7 @@ export const samplerLineSchema = z.strictObject({ export const udpLineSchema = z.strictObject({ t: timestampSchema, - host: z.enum(["uut", "loadgen"]), + host: samplerHostSchema, RcvbufErrors: count, SndbufErrors: count, InErrors: count diff --git a/src/analyze/types.ts b/src/analyze/types.ts index c1b878d..38b6356 100644 --- a/src/analyze/types.ts +++ b/src/analyze/types.ts @@ -62,11 +62,17 @@ export interface SippResults { calls: Call[]; } +/** + * A host of the sampler's files: the unit under test, the first load + * generator, or one of the other load generators (`worker-1`, ...). + */ +export type SamplerHost = "uut" | "loadgen" | `worker-${number}`; + /** One sampler reading of one process name, with duplicates summed. */ export interface ProcessSample { t: string; tMs: number; - host: "uut" | "loadgen"; + host: SamplerHost; name: string; pid: number; rssKB: number; @@ -78,7 +84,7 @@ export interface ProcessSample { /** One UDP counters reading. `drops` is RcvbufErrors + SndbufErrors. */ export interface UdpSample { tMs: number; - host: "uut" | "loadgen"; + host: SamplerHost; drops: number; } diff --git a/src/orchestrate/buildReport.ts b/src/orchestrate/buildReport.ts index 6f08777..2988f5d 100644 --- a/src/orchestrate/buildReport.ts +++ b/src/orchestrate/buildReport.ts @@ -64,7 +64,10 @@ export function buildReport( loadgen: { size: local ? "local" : plan.loadgen.size, vcpus: state.machines.loadgen.vcpus, - memoryMB: state.machines.loadgen.memoryMB + memoryMB: state.machines.loadgen.memoryMB, + ...(state.hosts.workers.length > 0 + ? { count: state.hosts.workers.length + 1 } + : {}) }, os: state.machines.uut.os, docker: state.machines.uut.docker, diff --git a/src/orchestrate/buildSippRoles.ts b/src/orchestrate/buildSippRoles.ts index 36a09b5..32f466f 100644 --- a/src/orchestrate/buildSippRoles.ts +++ b/src/orchestrate/buildSippRoles.ts @@ -2,15 +2,20 @@ * Copyright (C) 2026 by Fonoster Inc (https://fonoster.com). MIT License. */ import { durationToSeconds } from "../analyze/durationToSeconds.js"; -import { type BenchmarkConfig, targetOf } from "../schemas/index.js"; +import { + type BenchmarkConfig, + calleeInstancesOf, + targetOf +} from "../schemas/index.js"; +import { splitRate } from "./splitRate.js"; import { CAPACITY_RECV_TIMEOUT_MS, callIdOf, REGISTER_RATE, SIPP_LANES, SIPP_MAX_SOCKET_CAP, - SIPP_PORTS, SIPP_SOCKET_MARGIN, + sippPortOf, type SippRole } from "./constants.js"; @@ -19,6 +24,11 @@ export function uacUsersFileOf(instance: number): string { return `users-uac-${instance}.csv`; } +/** Where a callee instance's share of the register users file is, in /raw. */ +export function registerUsersFileOf(instance: number): string { + return `users-register-${instance}.csv`; +} + /** Where the project is mounted in a SIPp container. */ export const SIPP_PROJECT_DIR = "/project"; @@ -46,7 +56,7 @@ export const SIPP_COMMON_FLAGS = [ export interface SippRoleSpec { role: SippRole; - /** Which of the role's SIPp processes: 1, or 1 to `load.instances` for the uac. */ + /** Which of the role's SIPp processes: 1, or 1 to `load.instances` for the uac and the callees. */ instance: number; /** The scenario, relative to the project directory. */ scenario: string; @@ -61,7 +71,7 @@ export interface SippRolesInput { uutIp: string; /** The load generator's address: SIPp binds to it (`-i`), so it is [local_ip]. */ loadgenIp: string; - /** Rows of the register users file, after its first line. */ + /** Rows of the register users file, after its first line; the callees split them. */ registerUsers: number; /** A cluster benchmark's entry; otherwise `:`. */ target?: { ip: string; port: number }; @@ -71,13 +81,17 @@ export interface SippRolesInput { bindIps?: Partial>; /** A uac instance's own address, when it runs on another load generator. */ uacIps?: Record; + /** A callee instance's own address, when it runs on another load generator. */ + calleeIps?: Record; } /** * The SIPp roles of a benchmark, in the order they start, with the * Contract's flags. Either kind can have callees: the register scenario, - * answering INVITEs with the uas scenario through `-oocsf`, on port 5070, - * or the uas alone on 5070; they start before the uac. Each role has its + * answering INVITEs with the uas scenario through `-oocsf`, or the uas alone + * on 5070; they start before the uac. With a register there are as many + * callee instances as `load.instances`, each registering its share of the + * users (`-inf /raw/users-register-.csv`) from its own port. Each role has its * own Call-ID prefix (`-cid_str --%u-%p@%s`) and binds to the load * generator's address (`-i`), which scenarios use as [local_ip]. The uac and * register roles send to the target; no role gets `-s`. The hold is `-d`. @@ -94,7 +108,8 @@ export function buildSippRoles(input: SippRolesInput): SippRoleSpec[] { const { benchmark } = input; const { load } = benchmark; const lane = SIPP_LANES[targetOf(benchmark)]; - const portOf = (role: SippRole) => SIPP_PORTS[role] + lane.portOffset; + const portOf = (role: SippRole, instance = 1) => + sippPortOf(role, instance) + lane.portOffset; const target = input.target ? `${input.target.ip}:${input.target.port}` : `${input.uutIp}:${load.port}`; @@ -106,11 +121,11 @@ export function buildSippRoles(input: SippRolesInput): SippRoleSpec[] { "-sf", `${SIPP_PROJECT_DIR}/${scenario}`, "-i", - (role === "uac" ? input.uacIps?.[instance] : undefined) ?? + (role === "uac" ? input.uacIps?.[instance] : input.calleeIps?.[instance]) ?? input.bindIps?.[role] ?? input.loadgenIp, "-p", - String(portOf(role) + instance - 1), + String(portOf(role, instance)), "-cid_str", callIdOf(role, lane.prefix, input.runId, instance), ...(tcp ? ["-t", "tn", ...SIPP_COMMON_FLAGS.slice(2)] : SIPP_COMMON_FLAGS) @@ -125,9 +140,18 @@ export function buildSippRoles(input: SippRolesInput): SippRoleSpec[] { ? ["-inf", `/raw/${uacUsersFileOf(instance)}`] : users(load.uac.users); const n = (value: number) => String(value); - /** An instance's share of a rate, or of a count of open calls. */ - const share = (value: number) => - String(Math.round((value / instances) * 1e4) / 1e4); + /** An instance's share of a rate: whole numbers that add up to the rate. */ + const share = (value: number, instance: number) => + String(splitRate(value, instances)[instance - 1]); + /** A starting rate's share: every instance must get some, or it sends nothing. */ + const startShare = (value: number, instance: number) => { + if (value < instances) { + throw new Error( + `${benchmark.id}: a starting rate of ${value} cannot be split among ${instances} SIPp instances; start at ${instances} or more` + ); + } + return share(value, instance); + }; const shareCalls = (calls: number) => Math.ceil(calls / instances); const instanceList = Array.from({ length: instances }, (_, i) => i + 1); /** `-l `, and over TCP `-max_socket` for them. */ @@ -142,36 +166,44 @@ export function buildSippRoles(input: SippRolesInput): SippRoleSpec[] { : []) ]; + const calleeCount = calleeInstancesOf(load); + const calleeList = Array.from({ length: calleeCount }, (_, i) => i + 1); + /** A callee instance's share of the register users, or all of them. */ + const registerUsersOf = (instance: number) => + shareSizeOf(input.registerUsers, calleeCount, instance); + const registerInput = (register: { users: string }, instance: number) => + calleeCount > 1 + ? ["-inf", `/raw/${registerUsersFileOf(instance)}`] + : users(register.users); + const callees = (rampCalls: number | null): SippRoleSpec[] => { const { register, uas } = load; if (register) { - return [ - { - role: "register", - instance: 1, - scenario: register.scenario, - args: [ - target, - ...scenarioFlags("register", register.scenario), - ...(uas ? ["-oocsf", `${SIPP_PROJECT_DIR}/${uas.scenario}`] : []), - ...users(register.users), - "-m", - n(input.registerUsers), - "-r", - String(REGISTER_RATE), - ...open(input.registerUsers + (rampCalls ?? 0)), - "-key", - "expires", - n(register.expires), - "-key", - "refresh_ms", - n((register.expires / 2) * 1000), - "-key", - "uas_port", - String(portOf("register")) - ] - } - ]; + return calleeList.map((instance): SippRoleSpec => ({ + role: "register", + instance, + scenario: register.scenario, + args: [ + target, + ...scenarioFlags("register", register.scenario, instance), + ...(uas ? ["-oocsf", `${SIPP_PROJECT_DIR}/${uas.scenario}`] : []), + ...registerInput(register, instance), + "-m", + n(registerUsersOf(instance)), + "-r", + String(REGISTER_RATE), + ...open(registerUsersOf(instance) + shareCalls(rampCalls ?? 0)), + "-key", + "expires", + n(register.expires), + "-key", + "refresh_ms", + n((register.expires / 2) * 1000), + "-key", + "uas_port", + String(portOf("register", instance)) + ] + })); } if (uas) { return [ @@ -216,15 +248,15 @@ export function buildSippRoles(input: SippRolesInput): SippRoleSpec[] { ...scenarioFlags("uac", load.uac.scenario, instance), ...uacUsers(instance), "-r", - share(start), + startShare(start, instance), "-rp", "1000", "-rate_increase", - share(step), + share(step, instance), "-rate_interval", n(durationToSeconds(every)), "-rate_max", - share(max), + share(max, instance), "-no_rate_quit", "-recv_timeout", n(CAPACITY_RECV_TIMEOUT_MS), @@ -247,7 +279,7 @@ export function buildSippRoles(input: SippRolesInput): SippRoleSpec[] { ...scenarioFlags("uac", load.uac.scenario, instance), ...uacUsers(instance), "-r", - share(cps), + startShare(cps, instance), "-rp", "1000", ...open(shareCalls(soakCalls())), @@ -257,6 +289,18 @@ export function buildSippRoles(input: SippRolesInput): SippRoleSpec[] { ]; } +/** + * How many of `rows` users the `instance`-th (from 1) of `instances` shares + * gets when they are dealt out round-robin, like `splitUsers` does. + */ +export function shareSizeOf( + rows: number, + instances: number, + instance: number +): number { + return Math.max(0, Math.floor((rows - instance + instances) / instances)); +} + /** Rows of a SIPp injection file: its lines after the first, not blank. */ export function countUserRows(text: string): number { return text diff --git a/src/orchestrate/constants.ts b/src/orchestrate/constants.ts index 5341a09..c2f2125 100644 --- a/src/orchestrate/constants.ts +++ b/src/orchestrate/constants.ts @@ -9,11 +9,30 @@ export const REMOTE_ROOT = "/opt/perfkit"; * 5070: the register scenario, with the uas scenario answering out-of-call * INVITEs (`-oocsf`) on the same socket, because Routr routes a call to * the address of the REGISTER's top Via. A soak with a uas and no register - * runs the uas alone on 5070. + * runs the uas alone on 5070. With `load.instances` above 1 there are as many + * callee processes, each on its own port (see `sippPortOf`). */ export const SIPP_PORTS = { uac: 5080, register: 5070, uas: 5070 } as const; export type SippRole = keyof typeof SIPP_PORTS; +/** + * The first port of the callee instances after the first, which keeps 5070: + * 5080 and up are the uac instances', so the callees' go well above them. + */ +export const SIPP_EXTRA_CALLEE_PORT = 5200; + +/** + * A SIPp role instance's local port, before the lane's offset: the uac's + * instances count up from 5080, the first callee is on 5070 and the other + * callees on 5200 and up. + */ +export function sippPortOf(role: SippRole, instance = 1): number { + if (role === "uac") return SIPP_PORTS.uac + instance - 1; + return instance > 1 + ? SIPP_EXTRA_CALLEE_PORT + instance - 2 + : SIPP_PORTS[role]; +} + /** The sampler's process name for each SIPp role (#2's sipp-uac and peers). */ export const SIPP_PROCESS = { uac: "sipp-uac", diff --git a/src/orchestrate/createRunBenchmark.ts b/src/orchestrate/createRunBenchmark.ts index 5199762..1ffb14b 100644 --- a/src/orchestrate/createRunBenchmark.ts +++ b/src/orchestrate/createRunBenchmark.ts @@ -7,6 +7,7 @@ import { durationToSeconds } from "../analyze/durationToSeconds.js"; import type { Now, Sleep } from "../infra/index.js"; import { type BenchmarkConfig, + calleeInstancesOf, type ComposeUutConfig, type HelmUutConfig, isHelmUut, @@ -32,6 +33,8 @@ import { toKubeSamplerLines } from "./toKubeSamplerLines.js"; import { countRegistered } from "./countRegistered.js"; import { buildSippRoles, + registerUsersFileOf, + shareSizeOf, splitUsers, uacUsersFileOf, countUserRows, @@ -136,8 +139,9 @@ export function createRunBenchmark(deps: AgentDeps, state: RunState) { const workers = state.hosts.workers; const generatorOf = (instance: number): Host => generatorFor(instance, workers.length); - const hostOf = (role: SippRole, instance: number): Host => - role === "uac" ? generatorOf(instance) : "loadgen"; + /** Every uac and callee instance runs on the generator its number maps to. */ + const hostOf = (_role: SippRole, instance: number): Host => + generatorOf(instance); const workerHosts = workers.map((_, index): Host => `worker-${index + 1}`); /** `ssh` to a worker's private address, from the load generator. */ const sshTo = (index: number) => @@ -224,31 +228,41 @@ export function createRunBenchmark(deps: AgentDeps, state: RunState) { const containers = { samplerUut: name(`${benchmark.id}-sampler-uut`), samplerLoadgen: name(`${benchmark.id}-sampler-loadgen`), + samplerWorker: (index: number) => + name(`${benchmark.id}-sampler-worker-${index}`), otelcol: name(`${benchmark.id}-otelcol`), sipp: (role: string, instance = 1) => name(`${benchmark.id}-sipp-${roleOf(role as SippRole, instance)}`) }; const instances = benchmark.load.instances; const uacInstances = Array.from({ length: instances }, (_, i) => i + 1); + const calleeInstances = Array.from( + { length: calleeInstancesOf(benchmark.load) }, + (_, i) => i + 1 + ); const onGenerator = (host: Host) => - uacInstances - .filter((instance) => generatorOf(instance) === host) - .map((instance) => containers.sipp("uac", instance)); - const allSipp = [ - ...(["register", "uas"] as SippRole[]).map((role) => - containers.sipp(role) - ), - ...onGenerator("loadgen") - ]; - /** A uac instance on a worker binds to that worker's own address. */ - const uacIps = Object.fromEntries( - uacInstances.flatMap((instance) => { - const worker = workerIndexOf(generatorOf(instance)); - if (worker === null) return []; - const address = workers[worker - 1]!; - return [[instance, onCluster ? address.publicIp : address.privateIp]]; - }) - ) as Record; + [ + ...uacInstances.map((instance) => ["uac", instance] as const), + ...calleeInstances.flatMap((instance) => [ + ["register", instance] as const, + ["uas", instance] as const + ]) + ] + .filter(([role, instance]) => hostOf(role, instance) === host) + .map(([role, instance]) => containers.sipp(role, instance)); + const allSipp = onGenerator("loadgen"); + /** An instance on a worker binds to that worker's own address. */ + const addressesOf = (list: number[]) => + Object.fromEntries( + list.flatMap((instance) => { + const worker = workerIndexOf(generatorOf(instance)); + if (worker === null) return []; + const address = workers[worker - 1]!; + return [[instance, onCluster ? address.publicIp : address.privateIp]]; + }) + ) as Record; + const uacIps = addressesOf(uacInstances); + const calleeIps = addressesOf(calleeInstances); if (instances > 1 && onCluster && cluster?.fixedIps) { throw new AgentFailure( `${benchmark.id} has load.instances ${instances}, which a local cluster's fixed SIPp addresses don't allow; run it with one instance or on DigitalOcean` @@ -272,7 +286,8 @@ export function createRunBenchmark(deps: AgentDeps, state: RunState) { ...(cluster && onCluster && cluster.fixedIps ? { bindIps: cluster.sippIps } : {}), - uacIps + uacIps, + calleeIps }); // Fails early, before anything starts, when the rate is missing. if (!onCluster) buildRoles(null); @@ -290,6 +305,16 @@ export function createRunBenchmark(deps: AgentDeps, state: RunState) { 12 * 60 * 1000 ); }; + /** Stops the samplers of the other load generators, so their files are whole. */ + const removeWorkerSamplers = () => + Promise.all( + workerHosts.map((host, position) => + run( + host, + removeContainersCommand([containers.samplerWorker(position + 1)]) + ) + ) + ); const removeUutSampler = () => onCluster ? run("loadgen", removeContainersCommand([containers.samplerUut])) @@ -297,8 +322,14 @@ export function createRunBenchmark(deps: AgentDeps, state: RunState) { const cleanup = async () => { await Promise.all( - workerHosts.map((host) => - run(host, removeContainersCommand(onGenerator(host))) + workerHosts.map((host, position) => + run( + host, + removeContainersCommand([ + ...onGenerator(host), + containers.samplerWorker(position + 1) + ]) + ) ) ); await run( @@ -351,6 +382,7 @@ export function createRunBenchmark(deps: AgentDeps, state: RunState) { ) ) ); + await removeWorkerSamplers(); await pullWorkers(rawDir, true); if (benchmark.telemetry.otlp) { await run( @@ -605,7 +637,11 @@ export function createRunBenchmark(deps: AgentDeps, state: RunState) { async function startSamplers( benchmark: BenchmarkConfig, rawDir: string, - containers: { samplerUut: string; samplerLoadgen: string }, + containers: { + samplerUut: string; + samplerLoadgen: string; + samplerWorker: (index: number) => string; + }, prefix: string ) { const start = async (host: Host, container: string, spec: string) => { @@ -630,7 +666,7 @@ export function createRunBenchmark(deps: AgentDeps, state: RunState) { ); if (out.code !== 0) { throw new AgentFailure( - `could not start the sampler on the ${host === "uut" ? "server under test" : "load generator"}: ${lastLine(out.stderr, `exit ${out.code}`)}` + `could not start the sampler on the ${host === "uut" ? "server under test" : host === "loadgen" ? "load generator" : host}: ${lastLine(out.stderr, `exit ${out.code}`)}` ); } }; @@ -657,10 +693,20 @@ export function createRunBenchmark(deps: AgentDeps, state: RunState) { buildSamplerSpec("uut", benchmark.uut.processes) ); } - await start( + // Every load generator samples its own SIPp processes (the names are one + // per instance, so the spec is the same on all of them) and UDP counters + const loadgenSpec = buildSamplerSpec( "loadgen", - containers.samplerLoadgen, - buildSamplerSpec("loadgen", [], prefix, benchmark.load.instances) + [], + prefix, + benchmark.load.instances, + calleeInstancesOf(benchmark.load) + ); + await start("loadgen", containers.samplerLoadgen, loadgenSpec); + await Promise.all( + workerHosts.map((host, position) => + start(host, containers.samplerWorker(position + 1), loadgenSpec) + ) ); } @@ -713,9 +759,10 @@ export function createRunBenchmark(deps: AgentDeps, state: RunState) { name: container(spec.role, spec.instance), image: SIPP_IMAGE, detach: true, - // Several uac processes each get their own pid, so their trace - // files (__.csv) don't overwrite one another - ...(benchmark.load.instances > 1 && spec.role === "uac" + // Several uac or callee processes each get their own pid, so their + // trace files (__.csv) don't overwrite one another + ...(benchmark.load.instances > 1 && + (spec.role === "uac" || calleeInstancesOf(benchmark.load) > 1) ? { pidHost: true } : {}), platform: "linux/amd64", @@ -743,31 +790,58 @@ export function createRunBenchmark(deps: AgentDeps, state: RunState) { const uacs = roles.filter(({ role }) => role === "uac"); const uac = uacs[0] as SippRoleSpec; const callees = roles.filter(({ role }) => role !== "uac"); - if (uacs.length > 1 && benchmark.load.uac.users) { - const file = await deps.fs.readFile( - `${paths.project}/${benchmark.load.uac.users}` - ); + const registers = callees.filter(({ role }) => role === "register"); + /** Writes each instance's share of a users file into /raw. */ + const writeShares = async ( + text: string, + instances: number, + fileOf: (instance: number) => string + ) => { const shares = (() => { try { - return splitUsers(file, uacs.length); + return splitUsers(text, instances); } catch (err) { throw new AgentFailure(`${benchmarkId}: ${(err as Error).message}`); } })(); await Promise.all( shares.map((share, index) => - deps.fs.writeFile(`${rawDir}/${uacUsersFileOf(index + 1)}`, share) + deps.fs.writeFile(`${rawDir}/${fileOf(index + 1)}`, share) ) ); + }; + if (uacs.length > 1 && benchmark.load.uac.users) { + await writeShares( + await deps.fs.readFile(`${paths.project}/${benchmark.load.uac.users}`), + uacs.length, + uacUsersFileOf + ); } - // Each worker gets its own instances' shares of the uac users + if (registers.length > 1 && benchmark.load.register) { + await writeShares( + await deps.fs.readFile( + `${paths.project}/${benchmark.load.register.users}` + ), + registers.length, + registerUsersFileOf + ); + } + // Each worker gets the shares of the instances it runs: callers and callees await Promise.all( workers.map((_, position) => { const index = position + 1; - const own = uacs - .filter(({ instance }) => generatorOf(instance) === `worker-${index}`) - .map(({ instance }) => uacUsersFileOf(instance)); - return benchmark.load.uac.users && own.length > 0 + const here = (spec: SippRoleSpec) => + generatorOf(spec.instance) === `worker-${index}`; + // A worker only runs instances 2 and up, so the shares exist + const own = [ + ...(benchmark.load.uac.users + ? uacs.filter(here).map(({ instance }) => uacUsersFileOf(instance)) + : []), + ...registers + .filter(here) + .map(({ instance }) => registerUsersFileOf(instance)) + ]; + return own.length > 0 ? pushToWorker(index, rawDir, own) : run(`worker-${index}`, `mkdir -p ${shellQuote(rawDir)}`); }) @@ -776,21 +850,47 @@ export function createRunBenchmark(deps: AgentDeps, state: RunState) { (previous, spec) => previous.then(() => startRole(spec)), Promise.resolve() ); - const register = callees.find(({ role }) => role === "register"); - if (register && registerUsers > 0) { + if (registers.length > 0 && registerUsers > 0) { + const biggest = shareSizeOf(registerUsers, registers.length, 1); const timeoutMs = - Math.ceil(registerUsers / REGISTER_RATE) * 1000 + REGISTER_GRACE_MS; - const registered = await waitForRegistrations( - rawDir, - register.scenario, - registerUsers, - deps.now() + timeoutMs - ); - if (registered < registerUsers) { - return { - startedMs: deps.now(), - aborted: `only ${registered} of ${registerUsers} callees registered within ${Math.round(timeoutMs / 1000)} s, so no call was placed` - }; + Math.ceil(biggest / REGISTER_RATE) * 1000 + REGISTER_GRACE_MS; + const deadline = deps.now() + timeoutMs; + const seconds = Math.round(timeoutMs / 1000); + if (registers.length === 1) { + const registered = await waitForRegistrations( + rawDir, + (registers[0] as SippRoleSpec).scenario, + registerUsers, + deadline + ); + if (registered < registerUsers) { + return { + startedMs: deps.now(), + aborted: `only ${registered} of ${registerUsers} callees registered within ${seconds} s, so no call was placed` + }; + } + } else { + const counts = await waitForCalleeRegistrations( + registers, + rawDir, + registerUsers, + deadline, + (instance) => container("register", instance) + ); + const short = counts.filter( + ({ registered, expected }) => registered < expected + ); + if (short.length > 0) { + return { + startedMs: deps.now(), + aborted: `${short + .map( + ({ instance, registered, expected }) => + `callee ${instance} registered ${registered} of ${expected}` + ) + .join(", ")} within ${seconds} s, so no call was placed` + }; + } } await deps.log(`[${benchmarkId}] ${registerUsers} callees registered`); } @@ -829,6 +929,75 @@ export function createRunBenchmark(deps: AgentDeps, state: RunState) { return waitForRegistrations(rawDir, scenario, users, deadline); } + /** + * Polls every callee instance's counts file until each shows its share of + * the users registered or the deadline passes; returns each one's last + * count. A callee's file is named after its pid: with several callees each + * runs in the host's pid namespace, so the pid is the container's own, and + * a worker's file is read on the worker. + */ + async function waitForCalleeRegistrations( + registers: SippRoleSpec[], + rawDir: string, + totalUsers: number, + deadline: number, + container: (instance: number) => string + ): Promise< + Array<{ instance: number; registered: number; expected: number }> + > { + const pids = new Map(); + const countOf = async (spec: SippRoleSpec): Promise => { + const host = hostOf(spec.role, spec.instance); + if (!pids.has(spec.instance)) { + const out = await run( + host, + `docker inspect -f '{{.State.Pid}}' ${shellQuote(container(spec.instance))}`, + 60_000 + ); + const pid = out.stdout.trim(); + if (out.code === 0 && /^\d+$/.test(pid)) pids.set(spec.instance, pid); + } + const pid = pids.get(spec.instance); + if (pid === undefined) return 0; + const base = basename(spec.scenario).replace(/\.xml$/i, ""); + const file = `${base}_${pid}_counts.csv`; + const text = + host === "loadgen" + ? await deps.fs.readFile(`${rawDir}/${file}`).catch(() => "") + : ( + await run( + host, + `cat ${shellQuote(`${rawDir}/${file}`)} 2>/dev/null; true`, + 60_000 + ) + ).stdout; + return countRegistered(text); + }; + const read = async () => + Promise.all( + registers.map(async (spec) => ({ + instance: spec.instance, + registered: await countOf(spec), + expected: shareSizeOf(totalUsers, registers.length, spec.instance) + })) + ); + const counts = await read(); + if ( + counts.every(({ registered, expected }) => registered >= expected) || + deps.now() >= deadline + ) { + return counts; + } + await deps.sleep(REGISTER_POLL_MS); + return waitForCalleeRegistrations( + registers, + rawDir, + totalUsers, + deadline, + container + ); + } + async function sippStartTime( rawDir: string, scenario: string, @@ -952,7 +1121,9 @@ export function createRunBenchmark(deps: AgentDeps, state: RunState) { : ""; const merge = async (kind: "sampler" | "udp") => { const parts = await Promise.all( - ["uut", "loadgen"].map((host) => read(`${kind}-${host}.jsonl`)) + ["uut", "loadgen", ...workerHosts].map((host) => + read(`${kind}-${host}.jsonl`) + ) ); await deps.fs.writeFile( `${rawDir}/${kind}.jsonl`, diff --git a/src/orchestrate/createStartRun.ts b/src/orchestrate/createStartRun.ts index d110bf4..de26974 100644 --- a/src/orchestrate/createStartRun.ts +++ b/src/orchestrate/createStartRun.ts @@ -111,20 +111,23 @@ export function createStartRun(deps: StartRunDeps) { "could not pull the sampler image on the server under test" ); } - // The other load generators only run SIPp's uac + // The other load generators run SIPp's uac and callees, and a sampler await Promise.all( workers.map((worker) => exec( worker, - shellJoin([ - "docker", - "pull", - "-q", - "--platform", - "linux/amd64", - SIPP_IMAGE - ]), - `could not pull SIPp's image on the ${worker}` + [ + shellJoin([ + "docker", + "pull", + "-q", + "--platform", + "linux/amd64", + SIPP_IMAGE + ]), + shellJoin(["docker", "pull", "-q", SAMPLER_IMAGE]) + ].join(" && "), + `could not pull the images on the ${worker}` ) ) ); @@ -190,7 +193,6 @@ export function createStartRun(deps: StartRunDeps) { await Promise.all( hosts.map(async (host) => { await remote.upload(host, project, paths.project); - if (workers.includes(host)) return; await exec( host, writeFileCommand(paths.sampler, SAMPLER_SCRIPT), diff --git a/src/orchestrate/evaluateCapacityStep.ts b/src/orchestrate/evaluateCapacityStep.ts index 14b445b..8ec15a8 100644 --- a/src/orchestrate/evaluateCapacityStep.ts +++ b/src/orchestrate/evaluateCapacityStep.ts @@ -7,10 +7,15 @@ import { measureCapacityStep, parseSamplerJsonl, parseUdpJsonl, + calleeProcessNames, readSippResults, uacProcessNames } from "../analyze/index.js"; -import type { CapacityBenchmarkConfig, LivePoint } from "../schemas/index.js"; +import { + calleeInstancesOf, + type CapacityBenchmarkConfig, + type LivePoint +} from "../schemas/index.js"; import { liveAnalyzeDeps } from "./liveAnalyzeDeps.js"; import type { AgentFs } from "./types.js"; @@ -65,6 +70,18 @@ export async function evaluateCapacityStep(input: { } const read = async (file: string) => names.includes(file) ? deps.readFile(join(rawDir, file)) : ""; + /** The first load generator's file and every worker's, as one text. */ + const readLoadgens = async (kind: "sampler" | "udp") => + ( + await Promise.all( + names + .filter((file) => + new RegExp(`^${kind}-(loadgen|worker-\\d+)\\.jsonl$`).test(file) + ) + .sort() + .map(read) + ) + ).join(""); const measured = measureCapacityStep({ offeredCps, window: capacityStepWindow({ @@ -75,12 +92,13 @@ export async function evaluateCapacityStep(input: { }), uac, samples: parseSamplerJsonl( - await read("sampler-loadgen.jsonl"), + await readLoadgens("sampler"), "sampler-loadgen.jsonl" ), - udp: parseUdpJsonl(await read("udp-loadgen.jsonl"), "udp-loadgen.jsonl"), + udp: parseUdpJsonl(await readLoadgens("udp"), "udp-loadgen.jsonl"), uutProcesses: [], uacProcesses: uacProcessNames(benchmark.load.instances), + calleeProcesses: calleeProcessNames(calleeInstancesOf(benchmark.load)), unit: benchmark.load.unit }); const { p99LatencyMs, failedCallsPct } = benchmark.thresholds; diff --git a/src/orchestrate/sampler.ts b/src/orchestrate/sampler.ts index 8964f5e..383f615 100644 --- a/src/orchestrate/sampler.ts +++ b/src/orchestrate/sampler.ts @@ -1,7 +1,10 @@ /** * Copyright (C) 2026 by Fonoster Inc (https://fonoster.com). MIT License. */ -import { uacProcessName } from "../analyze/measureCapacityStep.js"; +import { + calleeProcessName, + uacProcessName +} from "../analyze/measureCapacityStep.js"; import { roleOf, SAMPLE_INTERVAL_S, @@ -52,22 +55,28 @@ done * The sampler spec for one host: the UUT's processes from perfkit.yaml, or * the load generator's SIPp roles, matched by their Call-ID prefix flag * (with the lane's prefix, `k8s-` for a cluster benchmark). With several uac - * instances each is its own process name (`sipp-uac`, `sipp-uac-2`, ...), so - * the load generator's saturation is judged per SIPp process. + * or callee instances each is its own process name (`sipp-uac`, `sipp-uac-2`, + * `sipp-register-2`, ...), so the load generator's saturation is judged per + * SIPp process. */ export function buildSamplerSpec( host: Host, processes: ReadonlyArray<{ name: string; match: string }>, prefix = "", - uacInstances = 1 + uacInstances = 1, + calleeInstances = 1 ): string { const roles = (Object.keys(SIPP_PROCESS) as SippRole[]).flatMap((role) => - role === "uac" - ? Array.from({ length: uacInstances }, (_, i) => ({ - name: uacProcessName(i + 1), - role: roleOf(role, i + 1) - })) - : [{ name: SIPP_PROCESS[role], role }] + Array.from( + { length: role === "uac" ? uacInstances : calleeInstances }, + (_, i) => ({ + name: + role === "uac" + ? uacProcessName(i + 1) + : calleeProcessName(role, i + 1), + role: roleOf(role, i + 1) + }) + ) ); const entries = host === "uut" diff --git a/src/orchestrate/splitRate.ts b/src/orchestrate/splitRate.ts new file mode 100644 index 0000000..563b8e6 --- /dev/null +++ b/src/orchestrate/splitRate.ts @@ -0,0 +1,23 @@ +/** + * Copyright (C) 2026 by Fonoster Inc (https://fonoster.com). MIT License. + */ + +/** + * Splits a rate over `instances` SIPp processes in whole numbers that add up + * to the rate: 50 over 8 is six 6s and two 7s, the first instances taking the + * remainder. SIPp's `-rate_increase` only takes an integer, so a share of + * 6.25 would keep the instance from starting. A rate that is not a whole + * number (a soak's `cps: 7.5`) is split evenly, as `-r` takes decimals. + */ +export function splitRate(value: number, instances: number): number[] { + if (!Number.isInteger(value)) { + const each = Math.round((value / instances) * 1e4) / 1e4; + return Array.from({ length: instances }, () => each); + } + const base = Math.floor(value / instances); + const extra = value % instances; + return Array.from( + { length: instances }, + (_, i) => base + (i < extra ? 1 : 0) + ); +} diff --git a/src/orchestrate/toKubeSamplerLines.ts b/src/orchestrate/toKubeSamplerLines.ts index 549ad03..99fea45 100644 --- a/src/orchestrate/toKubeSamplerLines.ts +++ b/src/orchestrate/toKubeSamplerLines.ts @@ -37,7 +37,8 @@ interface NameState { * memory is the pods' working set summed (`rssKB`); CPU is their usage * summed and integrated over time into `cpuTicks` at 100 ticks/s, so CPU % * over a window comes out as on the server; `pid` is 1 plus the restarts of - * the name's containers seen so far, so a restart reads as one; `fds` is 0. + * the name's containers seen so far (a replaced pod counts as one), so a + * restart reads as a rise; `fds` is 0. * A tick with no pod, no usage yet or a line that does not parse (the last * one, half written) gives no line. */ @@ -67,8 +68,14 @@ export function toKubeSamplerLines(text: string): string { const first = state.lastMs === null; line.pods.forEach(({ uid, restarts }) => { const before = state.seen.get(uid); + // A pod that was not there before is a replacement: one restart, or + // more when its own containers had restarted too. const added = - before === undefined ? (first ? 0 : restarts) : restarts - before; + before === undefined + ? first + ? 0 + : Math.max(1, restarts) + : restarts - before; state.restarts += Math.max(0, added); state.seen.set(uid, restarts); }); diff --git a/src/orchestrate/workerFiles.ts b/src/orchestrate/workerFiles.ts index 759f3a9..8a7d418 100644 --- a/src/orchestrate/workerFiles.ts +++ b/src/orchestrate/workerFiles.ts @@ -16,10 +16,12 @@ export function generatorOf(instance: number, workers: number): Host { * The name a worker's SIPp file gets in the run's raw directory. SIPp names * its files after its pid (`__`), which two machines can * share, so the worker's number goes in front of the pid, padded past any - * real one: worker 1's pid 4242 becomes 100004242. Null for a file that is - * not SIPp's (the users shares). + * real one: worker 1's pid 4242 becomes 100004242. A worker's sampler files + * (`sampler-worker-1.jsonl`, `udp-worker-1.jsonl`) already carry its name and + * keep it. Null for any other file (the users shares, the sampler's spec). */ export function workerFileName(worker: number, name: string): string | null { + if (/^(sampler|udp)-worker-\d+\.jsonl$/.test(name)) return name; const match = /^(.+)_(\d+)_(.*)$/.exec(name); if (!match) return null; const [, base, pid, rest] = match; diff --git a/src/render/app/createLiveController.ts b/src/render/app/createLiveController.ts index 9fd0736..1b4caf4 100644 --- a/src/render/app/createLiveController.ts +++ b/src/render/app/createLiveController.ts @@ -2,7 +2,8 @@ * Copyright (C) 2026 by Fonoster Inc (https://fonoster.com). MIT License. */ /// -import { type Live, liveSchema, type RunEntry } from "../../schemas/index.js"; +import type { Live, RunEntry } from "../../schemas/index.js"; +import { parseLiveTolerant } from "./parseLiveTolerant.js"; import type { LiveViewContext } from "../live/types.js"; import type { LiveControllerDeps, LiveMode } from "./liveController.types.js"; @@ -40,8 +41,7 @@ export function createLiveController(deps: LiveControllerDeps) { try { const response = await deps.fetchLive(deps.liveUrl); if (response.status !== 200) return null; - const parsed = liveSchema.safeParse(JSON.parse(response.text)); - return parsed.success ? parsed.data : null; + return parseLiveTolerant(JSON.parse(response.text)); } catch { return null; } diff --git a/src/render/app/parseLiveTolerant.ts b/src/render/app/parseLiveTolerant.ts new file mode 100644 index 0000000..32c18ed --- /dev/null +++ b/src/render/app/parseLiveTolerant.ts @@ -0,0 +1,43 @@ +/** + * Copyright (C) 2026 by Fonoster Inc (https://fonoster.com). MIT License. + */ +import { type Live, liveSchema } from "../../schemas/index.js"; + +/** How many rounds of stripping unknown fields before giving up. */ +const MAX_ROUNDS = 8; + +/** + * Parses a live.json the way the page needs it: a field this app does not + * know, written by a newer perfkit, is left out instead of making the whole + * file invalid, so a run in progress still shows. Anything else wrong with + * the file (a missing or mistyped field, another schema id) is invalid, + * as with `liveSchema` itself. + * + * @returns The live snapshot without the unknown fields, or null + */ +export function parseLiveTolerant(value: unknown, round = 0): Live | null { + const parsed = liveSchema.safeParse(value); + if (parsed.success) return parsed.data; + const unknown = parsed.error.issues.flatMap((issue) => + issue.code === "unrecognized_keys" ? [issue] : [] + ); + if (unknown.length === 0 || unknown.length !== parsed.error.issues.length) { + return null; + } + if (round >= MAX_ROUNDS) return null; + unknown.forEach(({ path, keys }) => { + const holder = path.reduce( + (node, key) => + node !== null && typeof node === "object" + ? (node as Record)[key] + : undefined, + value + ); + if (holder !== null && typeof holder === "object") { + keys.forEach((key) => { + delete (holder as Record)[key]; + }); + } + }); + return parseLiveTolerant(value, round + 1); +} diff --git a/src/render/sections/renderSetup.ts b/src/render/sections/renderSetup.ts index 80e0f01..880dff2 100644 --- a/src/render/sections/renderSetup.ts +++ b/src/render/sections/renderSetup.ts @@ -12,11 +12,11 @@ import type { PageContext } from "../types.js"; */ export function renderSetup(ctx: PageContext): string { const { run } = ctx.report; - const machine = (m: typeof run.infra.uut) => - `${m.size} · ${m.vcpus} vCPU · ${formatNumber(m.memoryMB / 1024)} GB`; + const machine = (m: typeof run.infra.uut, count = 1) => + `${count > 1 ? `${count} × ` : ""}${m.size} · ${m.vcpus} vCPU · ${formatNumber(m.memoryMB / 1024)} GB`; const servers: Array<[string, string]> = [ ["Server under test", machine(run.infra.uut)], - ["Load generator", machine(run.infra.loadgen)], + ["Load generator", machine(run.infra.loadgen, run.infra.loadgen.count)], ["Region", `DigitalOcean ${run.infra.region}`], ["OS", `${run.infra.os} · Docker ${run.infra.docker}`] ]; diff --git a/src/schemas/calleeInstancesOf.ts b/src/schemas/calleeInstancesOf.ts new file mode 100644 index 0000000..d40c65a --- /dev/null +++ b/src/schemas/calleeInstancesOf.ts @@ -0,0 +1,15 @@ +/** + * Copyright (C) 2026 by Fonoster Inc (https://fonoster.com). MIT License. + */ + +/** + * How many callee SIPp processes a benchmark runs: as many as `instances` + * when it has a `register` scenario (each registers its share of the users), + * else one: a uas alone answers on one address and has nothing to split. + */ +export function calleeInstancesOf(load: { + instances: number; + register?: unknown; +}): number { + return load.register ? load.instances : 1; +} diff --git a/src/schemas/checkStartCoversInstances.ts b/src/schemas/checkStartCoversInstances.ts new file mode 100644 index 0000000..585b336 --- /dev/null +++ b/src/schemas/checkStartCoversInstances.ts @@ -0,0 +1,25 @@ +/** + * Copyright (C) 2026 by Fonoster Inc (https://fonoster.com). MIT License. + */ +import type { z } from "zod/v4"; + +/** + * Adds an issue at `load.rate.start` when a capacity benchmark starts below + * the number of SIPp instances: the rate is split among them in whole + * numbers, so an instance would get none. For use inside the capacity + * benchmark's `superRefine`. + */ +export function checkStartCoversInstances( + benchmark: { load: { instances: number; rate: { start: number } } }, + ctx: z.core.$RefinementCtx +): void { + const { instances, rate } = benchmark.load; + if (rate.start < instances) { + ctx.addIssue({ + code: "custom", + path: ["load", "rate", "start"], + message: `expected a starting rate of ${instances} or more, one for each of the ${instances} SIPp instances`, + input: rate.start + }); + } +} diff --git a/src/schemas/index.ts b/src/schemas/index.ts index d9301ef..9ca5496 100644 --- a/src/schemas/index.ts +++ b/src/schemas/index.ts @@ -25,6 +25,7 @@ export { type SoakRateConfig, type UutConfig } from "./perfkitConfig.schema.js"; +export { calleeInstancesOf } from "./calleeInstancesOf.js"; export { isHelmUut } from "./isHelmUut.js"; export { targetOf, type BenchmarkTarget } from "./targetOf.js"; export { uutProcessNames } from "./uutProcessNames.js"; diff --git a/src/schemas/perfkitConfig.schema.ts b/src/schemas/perfkitConfig.schema.ts index 7535793..9494621 100644 --- a/src/schemas/perfkitConfig.schema.ts +++ b/src/schemas/perfkitConfig.schema.ts @@ -12,6 +12,7 @@ import { checkClusterForHelm } from "./checkClusterForHelm.js"; import { checkFromCapacity } from "./checkFromCapacity.js"; import { checkUniqueIds } from "./checkUniqueIds.js"; import { projectLiveSchema } from "./report.schema.js"; +import { checkStartCoversInstances } from "./checkStartCoversInstances.js"; import { checkStepEveryMultipleOfSippInterval } from "./checkStepEveryMultipleOfSippInterval.js"; import { checkWarmupShorterThanStep } from "./checkWarmupShorterThanStep.js"; import { @@ -337,7 +338,8 @@ export const capacityBenchmarkConfigSchema = z thresholds: capacityThresholdsSchema }) .superRefine(checkWarmupShorterThanStep) - .superRefine(checkStepEveryMultipleOfSippInterval); + .superRefine(checkStepEveryMultipleOfSippInterval) + .superRefine(checkStartCoversInstances); export const soakBenchmarkConfigSchema = z.strictObject({ id: benchmarkIdSchema, diff --git a/src/schemas/report.schema.ts b/src/schemas/report.schema.ts index af02c4b..2c17623 100644 --- a/src/schemas/report.schema.ts +++ b/src/schemas/report.schema.ts @@ -75,7 +75,16 @@ const runInfraSchema = z.strictObject({ provider: z.enum(["digitalocean", "local"]), region: nonEmptyString, uut: machineSchema, - loadgen: machineSchema, + /** + * The load generator droplets: `count` of them (omitted when there is + * one), each of this size. + */ + loadgen: machineSchema.extend({ + count: positiveNumber + .int({ error: "expected a whole number" }) + .min(2, { error: "expected 2 or more; leave it out for one" }) + .optional() + }), os: nonEmptyString, docker: nonEmptyString, cluster: runClusterSchema.optional() diff --git a/test/analyze/builders.ts b/test/analyze/builders.ts index eb39603..a9757f9 100644 --- a/test/analyze/builders.ts +++ b/test/analyze/builders.ts @@ -192,7 +192,7 @@ export function rttCsv( } export interface SamplerSpec { - host: "uut" | "loadgen"; + host: "uut" | "loadgen" | `worker-${number}`; name: string; pid?: number; rssKB?: number; @@ -218,7 +218,7 @@ export function samplerLines(tMs: number, specs: SamplerSpec[]): string[] { export function udpLine( tMs: number, - host: "uut" | "loadgen", + host: "uut" | "loadgen" | `worker-${number}`, RcvbufErrors = 0 ): string { return JSON.stringify({ diff --git a/test/analyze/countRestarts.unit.test.ts b/test/analyze/countRestarts.unit.test.ts new file mode 100644 index 0000000..560e7b9 --- /dev/null +++ b/test/analyze/countRestarts.unit.test.ts @@ -0,0 +1,67 @@ +/** + * Copyright (C) 2026 by Fonoster Inc (https://fonoster.com). MIT License. + */ +import { expect } from "chai"; +import { countRestarts } from "../../src/analyze/countRestarts.js"; +import type { ProcessSample } from "../../src/analyze/types.js"; + +const sample = (tick: number, name: string, pid: number): ProcessSample => ({ + t: new Date(tick * 10_000).toISOString(), + tMs: tick * 10_000, + host: "uut", + name, + pid, + rssKB: 1, + cpuTicks: 0, + clkTck: 100, + fds: 0 +}); + +/** Ticks 0 to 9 of `a` with the given pids; null leaves the tick out. */ +const run = (pids: Array) => + pids.flatMap((pid, tick) => (pid === null ? [] : [sample(tick, "a", pid)])); + +describe("countRestarts", () => { + it("should count a pid change and a run of missing ticks once each on a server", () => { + // Arrange: a pid change, then a gap of two ticks + // (the ticks are the times any process was sampled, so "b" keeps them) + const samples = [ + ...run([1, 1, 2, 2, null, null, 2, 2]), + ...Array.from({ length: 8 }, (_, tick) => sample(tick, "b", 7)) + ]; + + // Act and assert + expect(countRestarts(samples, ["a"])).to.equal(2); + }); + + describe("from the pods' counters", () => { + it("should not count missing ticks as restarts", () => { + // Arrange + const samples = run([1, null, 1, null, null, 1, 1]); + + // Act and assert + expect(countRestarts(samples, ["a"], { fromCounters: true })).to.equal(0); + }); + + it("should count each rise of the pid, also across a gap", () => { + // Arrange: 1 to 2, a gap, then 2 to 4 + const samples = run([1, 2, null, 2, 4]); + + // Act and assert + expect(countRestarts(samples, ["a"], { fromCounters: true })).to.equal(3); + }); + + it("should add up the names", () => { + // Arrange + const samples = [ + ...run([1, 2]), + ...[sample(0, "b", 1), sample(1, "b", 2)] + ]; + + // Act and assert + expect( + countRestarts(samples, ["a", "b"], { fromCounters: true }) + ).to.equal(2); + }); + }); +}); diff --git a/test/analyze/createAnalyzeSoak.unit.test.ts b/test/analyze/createAnalyzeSoak.unit.test.ts index 76913bc..8bd262b 100644 --- a/test/analyze/createAnalyzeSoak.unit.test.ts +++ b/test/analyze/createAnalyzeSoak.unit.test.ts @@ -184,6 +184,67 @@ describe("createAnalyzeSoak", () => { expect(result.rules.at(-1)).to.include({ measured: 1, passed: false }); }); + describe("on the cluster", () => { + const helmConfig = () => { + const base = soakConfig(); + return { + ...base, + uut: { + helm: { + chart: "../ops/charts/connect", + values: "k8s/values.yaml", + entry: { service: "routr-edgeport-tcp", port: 5060 }, + pods: ["one", "edgeport", "postgres"].map((name) => ({ + name, + selector: `service=${name}` + })) + } + } + } as unknown as typeof base; + }; + + it("should not count gaps in the samples as restarts", async () => { + // Arrange: 29 gaps, as the metrics API left in a real run, and no pod restarted + const gap = (name: string, hours: number) => + Array.from({ length: 29 }, (_, i) => 0.05 + i * 0.06).some( + (from) => hours >= from && hours < from + 0.02 + ) && name !== "one"; + + // Act + const result = measured( + await analyze( + { processAt: (name, hours) => (gap(name, hours) ? null : {}) }, + helmConfig() + ) + ); + + // Assert + expect(result.rules.at(-1)).to.deep.equal({ + id: "restarts", + limit: 0, + measured: 0, + passed: true + }); + }); + + it("should count a rise of the pid, the pod counters' restarts", async () => { + // Act + const result = measured( + await analyze( + { + processAt: (name, hours) => ({ + pid: name === "edgeport" ? (hours > 1 ? 3 : 1) : 1 + }) + }, + helmConfig() + ) + ); + + // Assert + expect(result.rules.at(-1)).to.include({ measured: 2, passed: false }); + }); + }); + it("should fail when failed calls reach 0.1% after the warm-up", async () => { // Act const result = measured(await analyze({ failedPerRow: 60 })); diff --git a/test/analyze/measureCapacityStep.unit.test.ts b/test/analyze/measureCapacityStep.unit.test.ts index 4797a52..07eba24 100644 --- a/test/analyze/measureCapacityStep.unit.test.ts +++ b/test/analyze/measureCapacityStep.unit.test.ts @@ -3,6 +3,8 @@ */ import { expect } from "chai"; import { + calleeProcessName, + calleeProcessNames, measureCapacityStep, parseSamplerJsonl, parseUdpJsonl, @@ -14,7 +16,8 @@ import { RAW_DIR, samplerLines, START_MS, - STEP_MS + STEP_MS, + udpLine } from "./builders.js"; /** A callee SIPp at `pct`% of one core, sampled every 10 s over the step. */ @@ -29,13 +32,19 @@ function calleeAt(name: string, pct: number): string[] { async function step( spec: Parameters[0][number], extraSampler: string[] = [], - uacProcesses?: string[] + uacProcesses?: string[], + calleeProcesses?: string[], + extraUdp: string[] = [] ) { const files = capacityRaw([spec]); files["sampler.jsonl"] = [ (files["sampler.jsonl"] as string).trimEnd(), ...extraSampler ].join("\n"); + files["udp.jsonl"] = [ + (files["udp.jsonl"] as string).trimEnd(), + ...extraUdp + ].join("\n"); const uac = await readSippResults( fakeDeps(files), RAW_DIR, @@ -53,6 +62,7 @@ async function step( udp: parseUdpJsonl(files["udp.jsonl"] as string, "udp.jsonl"), uutProcesses: ["edgeport"], ...(uacProcesses ? { uacProcesses } : {}), + ...(calleeProcesses ? { calleeProcesses } : {}), unit: "calls" }); } @@ -110,6 +120,94 @@ describe("measureCapacityStep", () => { expect(fine.step.sippCpuPct).to.equal(70); }); + it("should judge each callee instance on its own core, and name the busy one", async () => { + // Arrange: three callees, the third at 86% of its core + const callees = calleeProcessNames(3); + + // Act + const measured = await step( + { offeredCps: 40, attempted: 2400 }, + calleeAt("sipp-register-3", 86), + undefined, + callees + ); + + // Assert + expect(measured.invalidReason).to.equal( + "load generator saturated at 40 calls/s (sipp-register-3 at 86% of one core)" + ); + }); + + it("should judge a process on another load generator the same way", async () => { + // Arrange: the second uac and the second callee run on worker-1 + const onWorker = (name: string, pct: number) => + Array.from({ length: 7 }, (_, tick) => + samplerLines(START_MS + tick * 10_000, [ + { + host: "worker-1", + name, + cpuTicks: Math.round(tick * pct * 10) + } + ]) + ).flat(); + + // Act + const uac = await step( + { offeredCps: 40, attempted: 2400 }, + onWorker("sipp-uac-2", 86), + ["sipp-uac", "sipp-uac-2"] + ); + const callee = await step( + { offeredCps: 40, attempted: 2400 }, + onWorker("sipp-register-2", 90), + undefined, + calleeProcessNames(2) + ); + + // Assert + expect(uac.invalidReason).to.equal( + "load generator saturated at 40 calls/s (SIPp instance sipp-uac-2 at 86% of one core)" + ); + expect(callee.invalidReason).to.equal( + "load generator saturated at 40 calls/s (sipp-register-2 at 90% of one core)" + ); + }); + + it("should mark a step invalid when another load generator drops UDP packets", async () => { + // Arrange: worker-1's counter rises by 25 during the step + const udp = [ + udpLine(START_MS, "worker-1", 100), + udpLine(START_MS + 30_000, "worker-1", 110), + udpLine(START_MS + 60_000, "worker-1", 125) + ]; + + // Act + const measured = await step( + { offeredCps: 40, attempted: 2400 }, + [], + undefined, + undefined, + udp + ); + + // Assert + expect(measured.step.loadgenUdpDrops).to.equal(25); + expect(measured.invalidReason).to.equal( + "load generator dropped UDP packets at 40 calls/s (25 buffer errors)" + ); + }); + + it("should name the callee processes of an instance, the first without a number", () => { + expect(calleeProcessName("register", 1)).to.equal("sipp-register"); + expect(calleeProcessName("uas", 3)).to.equal("sipp-uas-3"); + expect(calleeProcessNames(2)).to.deep.equal([ + "sipp-register", + "sipp-uas", + "sipp-register-2", + "sipp-uas-2" + ]); + }); + it("should measure a valid step", async () => { // Act const measured = await step({ diff --git a/test/analyze/parseUdpJsonl.unit.test.ts b/test/analyze/parseUdpJsonl.unit.test.ts index e874a3d..8dda4e7 100644 --- a/test/analyze/parseUdpJsonl.unit.test.ts +++ b/test/analyze/parseUdpJsonl.unit.test.ts @@ -28,6 +28,22 @@ describe("parseUdpJsonl", () => { }); }); + it("should accept the udp counters of another load generator", () => { + // Arrange + const text = JSON.stringify({ + t: new Date(START_MS).toISOString(), + host: "worker-2", + RcvbufErrors: 1, + SndbufErrors: 0, + InErrors: 1 + }); + + // Act and assert + expect(parseUdpJsonl(text, "udp.jsonl")).to.deep.equal([ + { tMs: START_MS, host: "worker-2", drops: 1 } + ]); + }); + describe("with invalid input", () => { it("should throw ValidationError for an unknown host", () => { // Arrange diff --git a/test/orchestrate/agent.unit.test.ts b/test/orchestrate/agent.unit.test.ts index deb73aa..73aa660 100644 --- a/test/orchestrate/agent.unit.test.ts +++ b/test/orchestrate/agent.unit.test.ts @@ -92,6 +92,35 @@ async function routrWithCapacityCallees(): Promise { return join(dir, "perfkit.yaml"); } +/** A register role's counts file with `registered` 200 OKs. */ +function registerCounts(registered: number): string { + return [ + "CurrentTime;ElapsedTime;0_REGISTER_Sent;2_401_Recv;3_REGISTER_Sent;5_200_Recv;", + `2026-11-03T14:12:10Z;00:00:10:000000;200;200;200;${registered};`, + "" + ].join("\n"); +} + +/** + * Answers `docker inspect` of the callee containers with a pid of their own + * (1001 for the first, 1002 for -register2, ...), as SIPp names its files + * after it when several callees run in the host's pid namespace. + */ +function calleePids(setup: Awaited>) { + const writer = setup.hosts.onDocker; + setup.hosts.onDocker = async (host, command, by) => { + const match = /docker inspect .*-sipp-register(\d*)'?\s*$/.exec(command); + if (match) { + return { + code: 0, + stdout: `${1000 + Number(match[1] || 1)}\n`, + stderr: "" + }; + } + return writer(host, command, by); + }; +} + /** A passing capacity ramp, and the register role's counts file. */ function capacityWithRegistrations(registered: number) { return { @@ -398,9 +427,16 @@ describe("createAgent", () => { "--benchmarks", "one-soak" ], - soakRaw({ warmupS: 1800, afterS: 21600 }), + { + ...soakRaw({ warmupS: 1800, afterS: 21600 }), + // 200 callees dealt to 3 callees: 67, 67 and 66 + "uac_register_1001_counts.csv": registerCounts(67), + "uac_register_1002_counts.csv": registerCounts(67), + "uac_register_1003_counts.csv": registerCounts(66) + }, true ); + calleePids(setup); await withCallees(setup); const callers = Array.from({ length: 9 }, (_, i) => `caller${i};pw${i}`); await writeFile( @@ -420,7 +456,27 @@ describe("createAgent", () => { expect(started("-sipp-uac2 ")).to.have.length(1); expect(started("-sipp-uac3 ")).to.have.length(1); expect(started("-sipp-uac2 ")[0]).to.include("--pid host"); + // one callee per instance, each with its own port and share of the callees + expect(started("-sipp-register ")).to.have.length(1); + expect(started("-sipp-register2 ")).to.have.length(1); + expect(started("-sipp-register3 ")).to.have.length(1); + expect(started("-sipp-register2 ")[0]).to.include("--pid host"); + expect(started("-sipp-register2 ")[0]).to.include("-p 5200 "); + expect(started("-sipp-register2 ")[0]).to.include( + "-inf /raw/users-register-2.csv" + ); + expect(started("-sipp-register ")[0]).to.include("-m 67 "); + expect(started("-sipp-register3 ")[0]).to.include("-m 66 "); const raw = join(setup.hosts.roots.loadgen, "raw", "one-soak"); + expect( + (await readFile(join(raw, "users-register-3.csv"), "utf8")) + .trim() + .split("\n") + .slice(1, 3) + ).to.deep.equal([ + "callee2;pass2;[authentication username=callee2 password=pass2]", + "callee5;pass5;[authentication username=callee5 password=pass5]" + ]); const share = async (n: number) => (await readFile(join(raw, `users-uac-${n}.csv`), "utf8")) .trim() @@ -445,6 +501,62 @@ describe("createAgent", () => { ).to.equal(true); }); + it("should abort without starting the uacs when one callee registers only part of its share", async () => { + // Arrange: 3 callees, the second one registers 10 of its 67 + const dir = await mkdtemp(join(tmpdir(), "perfkit-callees-")); + await cp(dirname(ROUTR_CONFIG), dir, { recursive: true }); + const yaml = await readFile(ROUTR_CONFIG, "utf8"); + await writeFile( + join(dir, "perfkit.yaml"), + yaml.replace( + "uac: { scenario: one-soak/uac_invite.xml, users: one-soak/callers.csv }", + "uac: { scenario: one-soak/uac_invite.xml, users: one-soak/callers.csv }\n instances: 3" + ) + ); + const setup = await detached( + [ + "run", + "--config", + join(dir, "perfkit.yaml"), + "--target-version", + "2.31.5", + "--benchmarks", + "one-soak" + ], + { + ...soakRaw({ warmupS: 1800, afterS: 21600 }), + "uac_register_1001_counts.csv": registerCounts(67), + "uac_register_1002_counts.csv": registerCounts(10), + "uac_register_1003_counts.csv": registerCounts(66) + }, + true + ); + calleePids(setup); + await withCallees(setup); + await writeFile( + join(setup.hosts.roots.loadgen, "project", "one-soak", "callers.csv"), + [ + "SEQUENTIAL", + ...Array.from({ length: 9 }, (_, i) => `caller${i};pw${i}`), + "" + ].join("\n") + ); + + // Act + const outcome = await setup.agent.run(setup.runId()); + + // Assert + expect(outcome).to.deep.equal({ status: "done" }); + const commands = agentCommands(setup).map(({ command }) => command); + expect( + commands.some((c) => c.includes("-sipp-uac ") && c.includes("run -d")) + ).to.equal(false); + const raw = join(setup.hosts.roots.loadgen, "raw", "one-soak"); + expect((await readFile(join(raw, "aborted"), "utf8")).trim()).to.equal( + "callee 2 registered 10 of 67 within 64 s, so no call was placed" + ); + }); + it("should abort a soak without starting its uac when the callees don't all register in time", async () => { // Arrange const setup = await detached( diff --git a/test/orchestrate/buildReport.unit.test.ts b/test/orchestrate/buildReport.unit.test.ts index 57da410..c508206 100644 --- a/test/orchestrate/buildReport.unit.test.ts +++ b/test/orchestrate/buildReport.unit.test.ts @@ -76,6 +76,45 @@ describe("buildReport", () => { region: "nyc3" }); expect(report.project.logoSvg).to.match(/^ { + // Arrange + const setup = await doSetup({ agentOnStart: false }); + setup.infra.registry = fakeRegistry(); + await perfkit( + [ + "run", + "--config", + ROUTR_CONFIG, + "--target-version", + "2.31.5", + "--detach" + ], + setup.infra + ); + const state = runStateSchema.parse( + JSON.parse( + await readFile(join(setup.hosts.roots.loadgen, "run.json"), "utf8") + ) + ); + const withWorkers = { + ...state, + hosts: { + ...state.hosts, + workers: [ + { publicIp: "203.0.113.5", privateIp: "10.0.0.5" }, + { publicIp: "203.0.113.6", privateIp: "10.0.0.6" } + ] + } + }; + + // Act + const report = buildReport(withWorkers, new Map(), "2026-11-03T15:00:00Z"); + + // Assert + expect(report.run.infra.loadgen.count).to.equal(3); }); }); diff --git a/test/orchestrate/buildSippRoles.unit.test.ts b/test/orchestrate/buildSippRoles.unit.test.ts index c9f39fe..6b28666 100644 --- a/test/orchestrate/buildSippRoles.unit.test.ts +++ b/test/orchestrate/buildSippRoles.unit.test.ts @@ -5,6 +5,8 @@ import { expect } from "chai"; import { buildSippRoles, countUserRows, + registerUsersFileOf, + shareSizeOf, splitUsers, uacUsersFileOf } from "../../src/orchestrate/buildSippRoles.js"; @@ -494,7 +496,10 @@ describe("buildSippRoles with several uac instances", () => { it("should bind an instance on another load generator to that machine's address", () => { // Arrange - const benchmark = capacityWithCallees({ instances: 3 }); + const benchmark = capacityWithCallees({ + instances: 3, + rate: { start: 3, step: 3, every: "60s", max: 60 } + }); // Act const roles = buildSippRoles({ @@ -531,7 +536,7 @@ describe("buildSippRoles with several uac instances", () => { // Assert const uacs = roles.filter(({ role }) => role === "uac"); expect(uacs.map(({ instance }) => instance)).to.deep.equal([1, 2, 3, 4]); - expect(roles.filter(({ role }) => role !== "uac")).to.have.length(1); + expect(roles.filter(({ role }) => role !== "uac")).to.have.length(4); uacs.forEach(({ args, instance }) => { const flags = flagsOf(args); expect(flags).to.include("-r 25 -rp 1000 -rate_increase 25"); @@ -561,11 +566,56 @@ describe("buildSippRoles with several uac instances", () => { // Assert expect(uacs).to.have.length(3); - uacs.forEach(({ args }) => expect(flagsOf(args)).to.include("-r 33.3333")); + expect(uacs.map(({ args }) => args[args.indexOf("-r") + 1])).to.deep.equal([ + "34", + "33", + "33" + ]); const cids = uacs.map(({ args }) => args[args.indexOf("-cid_str") + 1]); expect(new Set(cids).size).to.equal(3); }); + it("should split a ramp that does not divide evenly into whole numbers per instance", () => { + // Arrange: 50 over 8 is 6.25, which SIPp's -rate_increase rejects + const benchmark = capacityWithCallees({ + instances: 8, + rate: { start: 50, step: 50, every: "30s", max: 3000 } + }); + + // Act + const uacs = buildSippRoles(input(benchmark)).filter( + ({ role }) => role === "uac" + ); + + // Assert + const valuesOf = (flag: string) => + uacs.map(({ args }) => Number(args[args.indexOf(flag) + 1])); + const sum = (values: number[]) => values.reduce((a, b) => a + b, 0); + [ + ["-r", 50], + ["-rate_increase", 50], + ["-rate_max", 3000] + ].forEach(([flag, total]) => { + const values = valuesOf(flag as string); + expect(values.every(Number.isInteger), `${flag} whole`).to.equal(true); + expect(sum(values), `${flag} total`).to.equal(total); + }); + expect(valuesOf("-r")).to.deep.equal([7, 7, 6, 6, 6, 6, 6, 6]); + }); + + it("should throw when a soak's rate is below the number of instances", () => { + // Arrange + const benchmark = soakBenchmarkConfigSchema.parse({ + ...soakConfig(), + load: { ...soakConfig().load, instances: 3, rate: { cps: 2 } } + }); + + // Act and assert + expect(() => buildSippRoles(input(benchmark as never))).to.throw( + "cannot be split among 3 SIPp instances" + ); + }); + it("should keep the whole users file for a single instance", () => { // Arrange const benchmark = capacityWithCallees(); @@ -583,6 +633,142 @@ describe("buildSippRoles with several uac instances", () => { }); }); +describe("buildSippRoles with several callee instances", () => { + const roles = (instances: number, extra = {}) => { + const benchmark = capacityWithCallees({ + instances, + rate: { start: 100, step: 100, every: "60s", max: 400 } + }); + return buildSippRoles({ + benchmark, + runId: "RUN1", + uutIp: "10.0.0.2", + loadgenIp: "10.0.0.3", + registerUsers: 200, + ...extra + }); + }; + const calleesOf = (list: ReturnType) => + list.filter(({ role }) => role === "register"); + + it("should start one callee per instance, each with its own port, Call-ID and share of the users", () => { + // Act + const callees = calleesOf(roles(3)); + + // Assert + expect(callees.map(({ instance }) => instance)).to.deep.equal([1, 2, 3]); + const port = (args: string[]) => args[args.indexOf("-p") + 1]; + expect(callees.map(({ args }) => port(args))).to.deep.equal([ + "5070", + "5200", + "5201" + ]); + callees.forEach(({ args, instance }) => { + const flags = flagsOf(args); + expect(flags).to.include( + `-cid_str ${instance === 1 ? "register" : `register${instance}`}-run1-%u-%p@%s` + ); + expect(flags).to.include(`-inf /raw/${registerUsersFileOf(instance)}`); + expect(flags).to.include(`-key uas_port ${port(args)}`); + expect(flags).to.include("-oocsf /project/one-soak/uas_answer.xml"); + }); + // 200 users dealt to 3 callees: 67, 67 and 66 + expect( + callees.map(({ args }) => args[args.indexOf("-m") + 1]) + ).to.deep.equal(["67", "67", "66"]); + }); + + it("should keep the whole register users file for a single callee", () => { + // Act + const callees = calleesOf(roles(1)); + + // Assert + expect(callees).to.have.length(1); + expect(flagsOf(callees[0]!.args)).to.include( + "-inf /project/one-soak/callees.csv" + ); + expect(flagsOf(callees[0]!.args)).to.include("-m 200 "); + }); + + it("should bind a callee on another load generator to that machine's address", () => { + // Act + const callees = calleesOf(roles(3, { calleeIps: { 2: "10.0.0.9" } })); + + // Assert + const bound = (instance: number) => { + const { args } = callees.find((spec) => spec.instance === instance)!; + return args[args.indexOf("-i") + 1]; + }; + expect([bound(1), bound(2), bound(3)]).to.deep.equal([ + "10.0.0.3", + "10.0.0.9", + "10.0.0.3" + ]); + }); + + it("should keep the uas alone as one process when there is no register", () => { + // Arrange: a soak whose callees are a uas only + const base = soakConfig(); + const benchmark = soakBenchmarkConfigSchema.parse({ + ...base, + load: { + ...base.load, + register: undefined, + instances: 3, + rate: { cps: 100 } + } + }); + + // Act + const list = buildSippRoles({ + benchmark: benchmark as never, + runId: "RUN1", + uutIp: "10.0.0.2", + loadgenIp: "10.0.0.3", + registerUsers: 0 + }); + + // Assert + expect(list.filter(({ role }) => role === "uas")).to.have.length(1); + expect(list.filter(({ role }) => role === "register")).to.have.length(0); + }); + + it("should never give a callee a port a uac uses", () => { + // Act: 64 instances, the most there can be: 64 callees and 64 uacs + const list = roles(64); + const ports = list.map(({ args }) => args[args.indexOf("-p") + 1]); + + // Assert + expect(ports).to.have.length(128); + expect(new Set(ports).size).to.equal(128); + }); +}); + +describe("shareSizeOf", () => { + it("should count the rows each of the round-robin shares gets", () => { + expect([1, 2, 3].map((i) => shareSizeOf(200, 3, i))).to.deep.equal([ + 67, 67, 66 + ]); + expect([1, 2, 3, 4].map((i) => shareSizeOf(10, 4, i))).to.deep.equal([ + 3, 3, 2, 2 + ]); + expect(shareSizeOf(200, 1, 1)).to.equal(200); + }); + + it("should agree with splitUsers", () => { + // Arrange + const text = `SEQUENTIAL\n${Array.from({ length: 23 }, (_, i) => `u${i}`).join("\n")}\n`; + + // Act + const shares = splitUsers(text, 5).map((share) => countUserRows(share)); + + // Assert + expect(shares).to.deep.equal( + [1, 2, 3, 4, 5].map((i) => shareSizeOf(23, 5, i)) + ); + }); +}); + describe("splitUsers", () => { it("should deal the rows out round-robin, each share with the mode line", () => { // Act diff --git a/test/orchestrate/evaluateCapacityStep.unit.test.ts b/test/orchestrate/evaluateCapacityStep.unit.test.ts new file mode 100644 index 0000000..77f1035 --- /dev/null +++ b/test/orchestrate/evaluateCapacityStep.unit.test.ts @@ -0,0 +1,100 @@ +/** + * Copyright (C) 2026 by Fonoster Inc (https://fonoster.com). MIT License. + */ +import { expect } from "chai"; +import { evaluateCapacityStep } from "../../src/orchestrate/evaluateCapacityStep.js"; +import type { AgentFs } from "../../src/orchestrate/types.js"; +import { + capacityConfig, + capacityRaw, + RAW_DIR, + samplerLines, + START_MS, + udpLine +} from "../analyze/builders.js"; + +/** The raw directory as the agent sees it mid-run: one sampler file per load generator. */ +function liveRaw(extra: Record): Record { + const files = capacityRaw([{ offeredCps: 50 }]); + const named: Record = { + ...files, + "sampler-loadgen.jsonl": files["sampler.jsonl"] as string, + "udp-loadgen.jsonl": files["udp.jsonl"] as string, + ...extra + }; + delete named["sampler.jsonl"]; + delete named["udp.jsonl"]; + return named; +} + +const fsOf = (files: Record): AgentFs => ({ + readFile: async (path) => files[path.slice(RAW_DIR.length + 1)] as string, + readdir: async () => Object.keys(files), + writeFile: async () => {}, + appendFile: async () => {}, + mkdir: async () => {}, + rename: async () => {}, + exists: async () => true +}); + +const evaluate = (files: Record) => + evaluateCapacityStep({ + fs: fsOf(files), + rawDir: RAW_DIR, + benchmark: capacityConfig({}, {}), + index: 0, + startMs: START_MS + }); + +describe("evaluateCapacityStep", () => { + it("should leave a step valid when the load generators are idle", async () => { + // Act + const evaluation = await evaluate(liveRaw({})); + + // Assert + expect(evaluation.stop).to.equal(false); + expect(evaluation.line).to.match(/ok$/); + }); + + it("should read another load generator's sampler file and call a saturated process invalid", async () => { + // Arrange: sipp-uac (instance 1) is the only uac the config names, so the + // saturated one here is the first callee process, on worker-1 + const busy = Array.from({ length: 7 }, (_, tick) => + samplerLines(START_MS + tick * 10_000, [ + { host: "worker-1", name: "sipp-register", cpuTicks: tick * 900 } + ]) + ) + .flat() + .join("\n"); + + // Act + const evaluation = await evaluate( + liveRaw({ "sampler-worker-1.jsonl": `${busy}\n` }) + ); + + // Assert + expect(evaluation.stop).to.equal(true); + expect(evaluation.line).to.include( + "invalid, load generator saturated at 50 messages/s (sipp-register at 90% of one core)" + ); + }); + + it("should read another load generator's UDP counters", async () => { + // Arrange + const udp = [ + udpLine(START_MS + 10_000, "worker-1", 0), + udpLine(START_MS + 40_000, "worker-1", 7) + ].join("\n"); + + // Act + const evaluation = await evaluate( + liveRaw({ "udp-worker-1.jsonl": `${udp}\n` }) + ); + + // Assert + expect(evaluation.stop).to.equal(true); + expect(evaluation.line).to.include( + "load generator dropped UDP packets at 50 messages/s (7 buffer errors)" + ); + }); +}); diff --git a/test/orchestrate/helpers.unit.test.ts b/test/orchestrate/helpers.unit.test.ts index ff29371..d879d3b 100644 --- a/test/orchestrate/helpers.unit.test.ts +++ b/test/orchestrate/helpers.unit.test.ts @@ -158,6 +158,12 @@ describe("orchestration helpers", () => { ); }); + it("should name each callee instance's SIPp process apart", () => { + expect(buildSamplerSpec("loadgen", [], "", 1, 2)).to.equal( + "sipp-uac\tsipp .*-cid_str uac-\nsipp-register\tsipp .*-cid_str register-\nsipp-register-2\tsipp .*-cid_str register2-\nsipp-uas\tsipp .*-cid_str uas-\nsipp-uas-2\tsipp .*-cid_str uas2-\n" + ); + }); + it("should stage dist, package.json and only the production dependencies", async () => { // Arrange const pkg = await fakePackage("1.2.3"); diff --git a/test/orchestrate/splitRate.unit.test.ts b/test/orchestrate/splitRate.unit.test.ts new file mode 100644 index 0000000..5cf98cc --- /dev/null +++ b/test/orchestrate/splitRate.unit.test.ts @@ -0,0 +1,32 @@ +/** + * Copyright (C) 2026 by Fonoster Inc (https://fonoster.com). MIT License. + */ +import { expect } from "chai"; +import { splitRate } from "../../src/orchestrate/splitRate.js"; + +describe("splitRate", () => { + it("should split 50 over 8 into whole numbers that add up to 50", () => { + // Act + const shares = splitRate(50, 8); + + // Assert + expect(shares).to.deep.equal([7, 7, 6, 6, 6, 6, 6, 6]); + expect(shares.reduce((a, b) => a + b, 0)).to.equal(50); + }); + + it("should split evenly when the rate divides", () => { + expect(splitRate(400, 4)).to.deep.equal([100, 100, 100, 100]); + }); + + it("should give a share of 0 to the last instances when the rate is below the count", () => { + expect(splitRate(2, 4)).to.deep.equal([1, 1, 0, 0]); + }); + + it("should split a rate that is not a whole number evenly", () => { + expect(splitRate(7.5, 3)).to.deep.equal([2.5, 2.5, 2.5]); + }); + + it("should return the rate itself for one instance", () => { + expect(splitRate(120, 1)).to.deep.equal([120]); + }); +}); diff --git a/test/orchestrate/toKubeSamplerLines.unit.test.ts b/test/orchestrate/toKubeSamplerLines.unit.test.ts index 4b5580c..6c2cca9 100644 --- a/test/orchestrate/toKubeSamplerLines.unit.test.ts +++ b/test/orchestrate/toKubeSamplerLines.unit.test.ts @@ -86,6 +86,28 @@ describe("toKubeSamplerLines", () => { expect(countRestarts(samples, ["edgeport"])).to.equal(1); }); + it("should count a pod that replaced another as a restart", () => { + // Arrange: pod a is gone at 20 s and a new pod (a new uid) takes its place + const text = [ + line(0, [{ pod: "a", uid: "u1", restarts: 0, cpu: "1m", mem: "1Ki" }]), + line(10, [{ pod: "a", uid: "u1", restarts: 0, cpu: "1m", mem: "1Ki" }]), + line(20, [{ pod: "b", uid: "u2", restarts: 0, cpu: "1m", mem: "1Ki" }]), + line(30, [{ pod: "b", uid: "u2", restarts: 0, cpu: "1m", mem: "1Ki" }]) + ].join("\n"); + + // Act + const samples = parseSamplerJsonl( + toKubeSamplerLines(text), + "sampler.jsonl" + ); + + // Assert + expect(samples.map(({ pid }) => pid)).to.deep.equal([1, 1, 2, 2]); + expect( + countRestarts(samples, ["edgeport"], { fromCounters: true }) + ).to.equal(1); + }); + it("should give no line for a tick without usage, and skip a half-written last line", () => { // Arrange const text = [ diff --git a/test/orchestrate/workerFiles.unit.test.ts b/test/orchestrate/workerFiles.unit.test.ts index 8c1a6cd..960fb76 100644 --- a/test/orchestrate/workerFiles.unit.test.ts +++ b/test/orchestrate/workerFiles.unit.test.ts @@ -46,6 +46,17 @@ describe("workerFileName", () => { ); }); + it("should keep a worker's sampler files under their own name and skip the other files", () => { + expect(workerFileName(1, "sampler-worker-1.jsonl")).to.equal( + "sampler-worker-1.jsonl" + ); + expect(workerFileName(2, "udp-worker-2.jsonl")).to.equal( + "udp-worker-2.jsonl" + ); + expect(workerFileName(1, "sampler-worker-1.spec")).to.equal(null); + expect(workerFileName(1, "users-uac-2.csv")).to.equal(null); + }); + it("should never collide with a pid of the load generator itself", () => { // A Linux pid has at most 7 digits; a renamed one has 9 or more const pid = (name: string) => /_(\d+)_/.exec(name)![1]!; diff --git a/test/render/app/parseLiveTolerant.unit.test.ts b/test/render/app/parseLiveTolerant.unit.test.ts new file mode 100644 index 0000000..661abae --- /dev/null +++ b/test/render/app/parseLiveTolerant.unit.test.ts @@ -0,0 +1,53 @@ +/** + * Copyright (C) 2026 by Fonoster Inc (https://fonoster.com). MIT License. + */ +import { expect } from "chai"; +import { parseLiveTolerant } from "../../../src/render/app/parseLiveTolerant.js"; +import { sampleLive } from "../../live/sample.js"; + +interface Snapshot { + [key: string]: unknown; + benchmarks: Array<{ + kind: string; + thresholds: Record; + points: Array>; + }>; +} + +const plain = (value: unknown): Snapshot => JSON.parse(JSON.stringify(value)); + +describe("parseLiveTolerant", () => { + it("should return a valid snapshot as it is", () => { + // Act and assert + expect(parseLiveTolerant(plain(sampleLive()))).to.deep.equal(sampleLive()); + }); + + it("should leave out unknown fields at any depth, so a newer perfkit's snapshot still shows", () => { + // Arrange: a new field at the top, in the thresholds and in a point + const live = plain(sampleLive()); + live.loadgens = 2; + const capacity = live.benchmarks.find(({ kind }) => kind === "capacity")!; + capacity.thresholds.confirmMore = 3; + capacity.points[0]!.extra = "x"; + + // Act + const parsed = parseLiveTolerant(live); + + // Assert + expect(parsed).to.deep.equal(sampleLive()); + }); + + it("should reject a snapshot with a wrong field, even when it has unknown ones too", () => { + // Arrange + const live = plain(sampleLive()); + live.loadgens = 2; + live.status = "paused"; + + // Act and assert + expect(parseLiveTolerant(live)).to.equal(null); + expect(parseLiveTolerant({})).to.equal(null); + expect(parseLiveTolerant({ ...plain(sampleLive()), schema: "x" })).to.equal( + null + ); + }); +}); diff --git a/test/render/renderRunPage.unit.test.ts b/test/render/renderRunPage.unit.test.ts index 4509ba8..7bab373 100644 --- a/test/render/renderRunPage.unit.test.ts +++ b/test/render/renderRunPage.unit.test.ts @@ -41,6 +41,29 @@ describe("the report page", () => { }); }); + describe("given a report with several load generators", () => { + it("should list their count in the test setup", async () => { + // Arrange + const report = loadReport("passed") as { + run: { infra: { loadgen: Record } }; + }; + report.run.infra.loadgen.count = 2; + const site = await renderSite(report, loadHistory(RUNS_4)); + const single = await renderSite( + loadReport("passed"), + loadHistory(RUNS_4) + ); + + // Act + const setup = textOf(site.latest(), ".setup"); + const alone = textOf(single.latest(), ".setup"); + + // Assert + expect(setup).to.match(/Load generator\s*2 × /); + expect(alone).to.not.match(/Load generator\s*\d+ × /); + }); + }); + describe("given the passed fixture with history", () => { it("should show the verdict, two Passed pills, the date chip and the takeaway", async () => { // Arrange diff --git a/test/schemas/perfkitConfig.unit.test.ts b/test/schemas/perfkitConfig.unit.test.ts index f794cbd..4ca879d 100644 --- a/test/schemas/perfkitConfig.unit.test.ts +++ b/test/schemas/perfkitConfig.unit.test.ts @@ -244,6 +244,15 @@ describe("perfkitConfigSchema", () => { ]); }); + it("should reject a capacity start below the number of SIPp instances", () => { + const raw = loadConfig(); + raw.benchmarks[0].load.instances = 8; + raw.benchmarks[0].load.rate.start = 5; + expect(errorsOf(perfkitConfigSchema, raw)).to.deep.equal([ + "benchmarks[0].load.rate.start: expected a starting rate of 8 or more, one for each of the 8 SIPp instances" + ]); + }); + it("should require duration and warmup in a soak benchmark", () => { const raw = loadConfig(); delete raw.benchmarks[1].duration;