From 236d9668b98f06eb70b385afa22ffc3798f0bbab Mon Sep 17 00:00:00 2001 From: Pedro Sanders Date: Mon, 9 Nov 2026 06:06:00 -0400 Subject: [PATCH 1/3] feat: load.instances splits a benchmark's uac over several SIPp processes SIPp is single-threaded, so one uac tops out at about one core (350 calls/s over TCP) however big the load generator is. load.instances: N starts N uac processes, each with 1/N of the rate and the open calls, its own port, Call-ID prefix and share of uac.users. The sampler names each process, the step check judges the busiest one, and the analysis sums them all. infra.loadgen.count is left as a design note. Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_01WSt25UoKGj2mgDNQ4dHVrB --- docs/design-loadgen-count.md | 42 ++++++++ docs/perfkit-yaml.md | 10 ++ schemas/perfkit-v1.schema.json | 12 +++ src/analyze/createAnalyzeCapacity.ts | 4 +- src/analyze/index.ts | 4 +- src/analyze/measureCapacityStep.ts | 39 ++++++-- src/analyze/readSippResults.ts | 5 +- src/analyze/sumStatRows.ts | 21 +++- src/analyze/types.ts | 2 + src/orchestrate/buildSippRoles.ts | 80 +++++++++++---- src/orchestrate/constants.ts | 18 +++- src/orchestrate/createRunBenchmark.ts | 66 ++++++++++--- src/orchestrate/evaluateCapacityStep.ts | 4 +- src/orchestrate/sampler.ts | 27 ++++- src/schemas/perfkitConfig.schema.ts | 11 +++ test/analyze/measureCapacityStep.unit.test.ts | 29 +++++- test/analyze/windows.unit.test.ts | 23 +++++ test/orchestrate/agent.unit.test.ts | 68 +++++++++++++ test/orchestrate/buildSippRoles.unit.test.ts | 99 ++++++++++++++++++- test/orchestrate/helpers.unit.test.ts | 9 ++ test/schemas/perfkitConfig.unit.test.ts | 14 +++ 21 files changed, 534 insertions(+), 53 deletions(-) create mode 100644 docs/design-loadgen-count.md diff --git a/docs/design-loadgen-count.md b/docs/design-loadgen-count.md new file mode 100644 index 0000000..202745e --- /dev/null +++ b/docs/design-loadgen-count.md @@ -0,0 +1,42 @@ +# Design: several load generator droplets (`infra.loadgen.count`) + +Status: not implemented. `load.instances` (SIPp processes on one load generator) ships first. + +## Why + +One load generator address has about 60,000 TCP source ports per destination, so over TCP it +caps at `cps × (hold + TIME_WAIT) ≈ 60,000` (for example 500 calls/s with 120 s holds). A very +large run (a 100M-call challenge: 1,000 to 3,000 calls/s for 10 hours or more) needs more +addresses, and several load generators also give the load balancer more than one source IP. + +## Shape + +```yaml +infra: + loadgen: { size: c-16, count: 4 } # default 1 +``` + +- `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. +- 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. + +## What it touches + +- `src/infra`: create, tag, wait for and sweep N droplets (the sweep already works by tag and + needs no change), the VPC and firewall rules for all of them, `RunState.hosts.loadgenIp` + becoming a list. +- `src/orchestrate`: the agent's `Host` type has `"loadgen"` hard-wired (`run("loadgen", ...)` + in `createRunBenchmark`, the sampler start, `collect`); it needs a worker address and a way to + fetch each worker's raw directory (the `uut` file fetch in `collectUutFiles` is the model). +- Live page and the report need nothing: they already sum the merged files. + +Cost: N × the load generator's hourly price; a c-16 is about $0.36/h, so `count: 4` adds about +$1.1/h over one. diff --git a/docs/perfkit-yaml.md b/docs/perfkit-yaml.md index eeb941b..a733122 100644 --- a/docs/perfkit-yaml.md +++ b/docs/perfkit-yaml.md @@ -151,6 +151,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.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. | @@ -308,3 +309,12 @@ benchmarks: telemetry: { otlp: true, sampleRatio: 0.01 } thresholds: { memoryGrowthMBPerHour: 5, failedCallsPct: 0.1, restarts: 0 } ``` + +### Scaling the load generator + +One c-16 load generator with `load.instances: 12` places about 12 times what one SIPp does. +Beyond one machine the limits are network and source ports: over TCP each call holds a +connection (`ip_local_port_range` is 1024-65535 with `tcp_tw_reuse`), so about 60,000 +connections per load generator address, which is `cps × (hold + TIME_WAIT)`. Several load +generator droplets (`infra.loadgen.count`) are the next step and are not implemented yet; see +the design in `docs/design-loadgen-count.md`. diff --git a/schemas/perfkit-v1.schema.json b/schemas/perfkit-v1.schema.json index bc07235..929182e 100644 --- a/schemas/perfkit-v1.schema.json +++ b/schemas/perfkit-v1.schema.json @@ -396,6 +396,12 @@ ], "additionalProperties": false }, + "instances": { + "default": 1, + "type": "integer", + "minimum": 1, + "maximum": 64 + }, "hold": { "default": "0s", "type": "string", @@ -734,6 +740,12 @@ ], "additionalProperties": false }, + "instances": { + "default": 1, + "type": "integer", + "minimum": 1, + "maximum": 64 + }, "hold": { "default": "0s", "type": "string", diff --git a/src/analyze/createAnalyzeCapacity.ts b/src/analyze/createAnalyzeCapacity.ts index 1070fc7..a749bfa 100644 --- a/src/analyze/createAnalyzeCapacity.ts +++ b/src/analyze/createAnalyzeCapacity.ts @@ -16,7 +16,8 @@ import { computeHops } from "./computeHops.js"; import { listRawDir } from "./listRawDir.js"; import { type MeasuredStep, - measureCapacityStep + measureCapacityStep, + uacProcessNames } from "./measureCapacityStep.js"; import { parseJsonDocument } from "./parseJsonDocument.js"; import { parseOtlpSpanLines } from "./parseOtlpSpans.js"; @@ -85,6 +86,7 @@ export function createAnalyzeCapacity(deps: AnalyzeDeps) { samples, udp, uutProcesses, + uacProcesses: uacProcessNames(benchmark.load.instances), unit }) ); diff --git a/src/analyze/index.ts b/src/analyze/index.ts index ae5e14b..6f9cf9a 100644 --- a/src/analyze/index.ts +++ b/src/analyze/index.ts @@ -33,7 +33,9 @@ export { MAX_SIPP_CPU_PCT, measureCapacityStep, MIN_ACHIEVED_RATIO, - SIPP_UAC_PROCESS + SIPP_UAC_PROCESS, + uacProcessName, + uacProcessNames } from "./measureCapacityStep.js"; export { nearestRank } from "./nearestRank.js"; export { parseOtlpSpanLines, parseOtlpSpans } from "./parseOtlpSpans.js"; diff --git a/src/analyze/measureCapacityStep.ts b/src/analyze/measureCapacityStep.ts index e599292..2ce474d 100644 --- a/src/analyze/measureCapacityStep.ts +++ b/src/analyze/measureCapacityStep.ts @@ -26,6 +26,21 @@ export const MIN_ACHIEVED_RATIO = 0.95; export const MAX_ACHIEVED_RATIO = 1.05; /** The load generator's SIPp process, as the sampler names it. */ export const SIPP_UAC_PROCESS = "sipp-uac"; + +/** + * The sampler's name for one SIPp uac instance: `sipp-uac` for the first, + * `sipp-uac-2` and on for the others. + */ +export function uacProcessName(instance: number): string { + return instance > 1 ? `${SIPP_UAC_PROCESS}-${instance}` : SIPP_UAC_PROCESS; +} + +/** The sampler names of `instances` uac processes, in order. */ +export function uacProcessNames(instances: number): string[] { + return Array.from({ length: Math.max(1, instances) }, (_, index) => + uacProcessName(index + 1) + ); +} /** The callee SIPp processes, checked against the same 80%. */ export const SIPP_CALLEE_PROCESSES = ["sipp-register", "sipp-uas"] as const; @@ -36,6 +51,8 @@ export interface StepInput { samples: readonly ProcessSample[]; udp: readonly UdpSample[]; uutProcesses: readonly string[]; + /** The uac SIPp processes' sampler names; just `sipp-uac` by default. */ + uacProcesses?: readonly string[]; unit: string; } @@ -50,7 +67,7 @@ export interface MeasuredStep { * Measures one capacity step. Latency counts the calls whose request was * sent in the step; attempts, failures and retransmissions come from the * stat rows in it. A step is invalid when the uac SIPp, or a callee SIPp - * (register, uas), used more than 80% of one core, placed less than 95% or more than 105% of the offered rate, or the load generator's + * (register, uas), used more than 80% of one core (any one of several uac instances), placed less than 95% or more than 105% of the offered rate, or the load generator's * UDP drops rose. */ export function measureCapacityStep(input: StepInput): MeasuredStep { @@ -64,12 +81,20 @@ export function measureCapacityStep(input: StepInput): MeasuredStep { const failedPct = roundTo( sums.attempted > 0 ? (sums.failed / sums.attempted) * 100 : 0 ); - const sippCpuPct = roundTo( - cpuPctOver( - processSamples(input.samples, "loadgen", SIPP_UAC_PROCESS), - window - ) + // Each instance is its own process and has its own core to saturate, so the + // busiest one decides, not the sum + const instanceCpu = (input.uacProcesses ?? [SIPP_UAC_PROCESS]).map( + (name) => ({ + name, + cpuPct: cpuPctOver(processSamples(input.samples, "loadgen", name), window) + }) + ); + const busiest = instanceCpu.reduce((top, one) => + one.cpuPct > top.cpuPct ? one : top ); + const sippCpuPct = roundTo(busiest.cpuPct); + const sippWhat = + instanceCpu.length > 1 ? `SIPp instance ${busiest.name}` : "SIPp"; const uutCpuPct = roundTo( input.uutProcesses.reduce( (sum, name) => @@ -86,7 +111,7 @@ export function measureCapacityStep(input: StepInput): MeasuredStep { const rate = `${formatNumber(offeredCps)} ${input.unit}/s`; const invalidReason = sippCpuPct > MAX_SIPP_CPU_PCT - ? `load generator saturated at ${rate} (SIPp at ${Math.round(sippCpuPct)}% of one core)` + ? `load generator saturated at ${rate} (${sippWhat} at ${Math.round(sippCpuPct)}% of one core)` : callee ? `load generator saturated at ${rate} (${callee.name} at ${Math.round(callee.cpuPct)}% of one core)` : achievedCps < offeredCps * MIN_ACHIEVED_RATIO diff --git a/src/analyze/readSippResults.ts b/src/analyze/readSippResults.ts index f2845bc..d0a2541 100644 --- a/src/analyze/readSippResults.ts +++ b/src/analyze/readSippResults.ts @@ -47,7 +47,10 @@ export async function readSippResults( sentMs: stat.startMs + row.atMs - row.responseTimeMs, latencyMs: row.responseTimeMs })); - return { rows: stat.rows, calls }; + return { + rows: stat.rows.map((row) => ({ ...row, pid })), + calls + }; }) ); return { diff --git a/src/analyze/sumStatRows.ts b/src/analyze/sumStatRows.ts index ab0ffa4..5caead2 100644 --- a/src/analyze/sumStatRows.ts +++ b/src/analyze/sumStatRows.ts @@ -13,7 +13,10 @@ export interface StatSums { failuresByReason: Record; /** Retransmissions(P). */ retransmissions: number; - /** The highest CurrentCall. */ + /** + * The highest CurrentCall; with several SIPp processes, the sum of each + * one's highest. + */ maxCurrentCalls: number; } @@ -51,9 +54,17 @@ export function sumStatRows( failed: sum("FailedCall(P)"), failuresByReason, retransmissions: sum("Retransmissions(P)"), - maxCurrentCalls: inside.reduce( - (max, row) => Math.max(max, row.counters.CurrentCall ?? 0), - 0 - ) + maxCurrentCalls: maxCurrentCalls(inside) }; } + +function maxCurrentCalls(rows: readonly StatRow[]): number { + const perProcess = rows.reduce((max, row) => { + const key = row.pid ?? ""; + return max.set( + key, + Math.max(max.get(key) ?? 0, row.counters.CurrentCall ?? 0) + ); + }, new Map()); + return [...perProcess.values()].reduce((sum, value) => sum + value, 0); +} diff --git a/src/analyze/types.ts b/src/analyze/types.ts index 613e643..c1b878d 100644 --- a/src/analyze/types.ts +++ b/src/analyze/types.ts @@ -29,6 +29,8 @@ export interface StatRow { periodStartMs: number; /** Every numeric column, by its header name, like `OutgoingCall(P)`. */ counters: Record; + /** The SIPp process that wrote the row, when several instances did. */ + pid?: string; } /** A SIPp `-trace_stat` file. */ diff --git a/src/orchestrate/buildSippRoles.ts b/src/orchestrate/buildSippRoles.ts index 1628636..9cb5fe8 100644 --- a/src/orchestrate/buildSippRoles.ts +++ b/src/orchestrate/buildSippRoles.ts @@ -14,6 +14,11 @@ import { type SippRole } from "./constants.js"; +/** Where an instance's share of the uac users file is, in a SIPp container. */ +export function uacUsersFileOf(instance: number): string { + return `users-uac-${instance}.csv`; +} + /** Where the project is mounted in a SIPp container. */ export const SIPP_PROJECT_DIR = "/project"; @@ -41,6 +46,8 @@ 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. */ + instance: number; /** The scenario, relative to the project directory. */ scenario: string; /** SIPp's arguments, after the binary. */ @@ -92,20 +99,33 @@ export function buildSippRoles(input: SippRolesInput): SippRoleSpec[] { const holdS = durationToSeconds(load.hold); const hold = ["-d", String(holdS * 1000)]; const tcp = load.transport === "tcp"; - const scenarioFlags = (role: SippRole, scenario: string) => [ + const instances = load.instances; + const scenarioFlags = (role: SippRole, scenario: string, instance = 1) => [ "-sf", `${SIPP_PROJECT_DIR}/${scenario}`, "-i", input.bindIps?.[role] ?? input.loadgenIp, "-p", - String(portOf(role)), + String(portOf(role) + instance - 1), "-cid_str", - callIdOf(role, lane.prefix, input.runId), + callIdOf(role, lane.prefix, input.runId, instance), ...(tcp ? ["-t", "tn", ...SIPP_COMMON_FLAGS.slice(2)] : SIPP_COMMON_FLAGS) ]; const users = (file: string | undefined) => file ? ["-inf", `${SIPP_PROJECT_DIR}/${file}`] : []; + /** The uac's users: the whole file, or an instance's share of it in /raw. */ + const uacUsers = (instance: number) => + load.uac.users === undefined + ? [] + : instances > 1 + ? ["-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); + 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. */ const open = (calls: number) => [ "-l", @@ -124,6 +144,7 @@ export function buildSippRoles(input: SippRolesInput): SippRoleSpec[] { return [ { role: "register", + instance: 1, scenario: register.scenario, args: [ target, @@ -152,6 +173,7 @@ export function buildSippRoles(input: SippRolesInput): SippRoleSpec[] { return [ { role: "uas", + instance: 1, scenario: uas.scenario, args: [ ...scenarioFlags("uas", uas.scenario), @@ -181,51 +203,53 @@ export function buildSippRoles(input: SippRolesInput): SippRoleSpec[] { const rampCalls = Math.ceil(max * (holdS + 20)); return [ ...callees(rampCalls), - { + ...instanceList.map((instance): SippRoleSpec => ({ role: "uac", + instance, scenario: load.uac.scenario, args: [ target, - ...scenarioFlags("uac", load.uac.scenario), - ...users(load.uac.users), + ...scenarioFlags("uac", load.uac.scenario, instance), + ...uacUsers(instance), "-r", - n(start), + share(start), "-rp", "1000", "-rate_increase", - n(step), + share(step), "-rate_interval", n(durationToSeconds(every)), "-rate_max", - n(max), + share(max), "-no_rate_quit", "-recv_timeout", n(CAPACITY_RECV_TIMEOUT_MS), - ...open(rampCalls), + ...open(shareCalls(rampCalls)), ...hold ] - } + })) ]; } const cps = soakCps(); return [ ...callees(null), - { + ...instanceList.map((instance): SippRoleSpec => ({ role: "uac", + instance, scenario: load.uac.scenario, args: [ target, - ...scenarioFlags("uac", load.uac.scenario), - ...users(load.uac.users), + ...scenarioFlags("uac", load.uac.scenario, instance), + ...uacUsers(instance), "-r", - n(cps), + share(cps), "-rp", "1000", - ...open(soakCalls()), + ...open(shareCalls(soakCalls())), ...hold ] - } + })) ]; } @@ -236,3 +260,25 @@ export function countUserRows(text: string): number { .slice(1) .filter((line) => line.trim() !== "").length; } + +/** + * Splits a SIPp injection file among `instances` uac processes: each gets the + * first line (the mode, like SEQUENTIAL) and every `instances`-th row, so a + * user is used by one instance only. Throws when there are fewer users than + * instances. + */ +export function splitUsers(text: string, instances: number): string[] { + const lines = text.split("\n"); + const header = lines[0] ?? ""; + const rows = lines.slice(1).filter((line) => line.trim() !== ""); + if (rows.length < instances) { + throw new Error( + `the users file has ${rows.length} users, fewer than the ${instances} SIPp instances` + ); + } + return Array.from( + { length: instances }, + (_, instance) => + `${header}\n${rows.filter((_row, index) => index % instances === instance).join("\n")}\n` + ); +} diff --git a/src/orchestrate/constants.ts b/src/orchestrate/constants.ts index f701fbc..5341a09 100644 --- a/src/orchestrate/constants.ts +++ b/src/orchestrate/constants.ts @@ -28,8 +28,22 @@ export const SIPP_PROCESS = { * samplers of a server soak and a cluster soak running at the same time * tell them apart. */ -export function callIdOf(role: SippRole, prefix = "", runId?: string): string { - return `${prefix}${role}-${runId ? `${runId.toLowerCase()}-` : ""}%u-%p@%s`; +export function callIdOf( + role: SippRole, + prefix = "", + runId?: string, + instance = 1 +): string { + return `${prefix}${roleOf(role, instance)}-${runId ? `${runId.toLowerCase()}-` : ""}%u-%p@%s`; +} + +/** + * A role's name for one of its instances: the first is the role itself + * (`uac`), the others carry their number (`uac2`), so every instance has a + * Call-ID prefix the sampler can tell apart (`uac-` never matches `uac2-`). + */ +export function roleOf(role: SippRole, instance = 1): string { + return instance > 1 ? `${role}${instance}` : role; } /** diff --git a/src/orchestrate/createRunBenchmark.ts b/src/orchestrate/createRunBenchmark.ts index 5f3078b..5b18933 100644 --- a/src/orchestrate/createRunBenchmark.ts +++ b/src/orchestrate/createRunBenchmark.ts @@ -32,6 +32,8 @@ import { toKubeSamplerLines } from "./toKubeSamplerLines.js"; import { countRegistered } from "./countRegistered.js"; import { buildSippRoles, + splitUsers, + uacUsersFileOf, countUserRows, SIPP_PROJECT_DIR, type SippRoleSpec @@ -58,6 +60,7 @@ import { SIPP_LANES, SIPP_START_TIMEOUT_MS, SIPP_TCP_NOFILE, + roleOf, type SippRole, STEP_GRACE_MS, UUT_UNRESPONSIVE @@ -149,9 +152,23 @@ export function createRunBenchmark(deps: AgentDeps, state: RunState) { samplerUut: name(`${benchmark.id}-sampler-uut`), samplerLoadgen: name(`${benchmark.id}-sampler-loadgen`), otelcol: name(`${benchmark.id}-otelcol`), - sipp: (role: string) => name(`${benchmark.id}-sipp-${role}`) + sipp: (role: string, instance = 1) => + name(`${benchmark.id}-sipp-${roleOf(role as SippRole, instance)}`) }; - const allRoles: SippRole[] = ["uac", "register", "uas"]; + const instances = benchmark.load.instances; + const allSipp = [ + ...(["register", "uas"] as SippRole[]).map((role) => + containers.sipp(role) + ), + ...Array.from({ length: instances }, (_, i) => + containers.sipp("uac", i + 1) + ) + ]; + 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` + ); + } const projectFile = (file: string) => deps.fs.readFile(`${paths.project}/${file}`); const registerUsers = benchmark.load.register @@ -196,7 +213,7 @@ export function createRunBenchmark(deps: AgentDeps, state: RunState) { await run( "loadgen", removeContainersCommand([ - ...allRoles.map((role) => containers.sipp(role)), + ...allSipp, containers.otelcol, containers.samplerLoadgen ]) @@ -235,8 +252,12 @@ export function createRunBenchmark(deps: AgentDeps, state: RunState) { await Promise.all( [...roles] .reverse() - .map(({ role }) => - run("loadgen", stopSippCommand(containers.sipp(role)), 120_000) + .map(({ role, instance }) => + run( + "loadgen", + stopSippCommand(containers.sipp(role, instance)), + 120_000 + ) ) ); if (benchmark.telemetry.otlp) { @@ -547,7 +568,7 @@ export function createRunBenchmark(deps: AgentDeps, state: RunState) { await start( "loadgen", containers.samplerLoadgen, - buildSamplerSpec("loadgen", [], prefix) + buildSamplerSpec("loadgen", [], prefix, benchmark.load.instances) ); } @@ -584,7 +605,7 @@ export function createRunBenchmark(deps: AgentDeps, state: RunState) { benchmark: BenchmarkConfig, roles: SippRoleSpec[], rawDir: string, - container: (role: string) => string, + container: (role: string, instance?: number) => string, registerUsers: number ): Promise<{ startedMs: number; aborted: string | null }> { const benchmarkId = benchmark.id; @@ -597,9 +618,14 @@ export function createRunBenchmark(deps: AgentDeps, state: RunState) { "loadgen", dockerRunCommand({ runId: state.runId, - name: container(spec.role), + 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" + ? { pidHost: true } + : {}), platform: "linux/amd64", ...(cluster ? { @@ -618,12 +644,30 @@ export function createRunBenchmark(deps: AgentDeps, state: RunState) { ); if (out.code !== 0) { throw new AgentFailure( - `could not start SIPp (${spec.role}): ${lastLine(out.stderr, `exit ${out.code}`)}` + `could not start SIPp (${roleOf(spec.role, spec.instance)}): ${lastLine(out.stderr, `exit ${out.code}`)}` ); } }; - const uac = roles.find(({ role }) => role === "uac") as SippRoleSpec; + 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 shares = (() => { + try { + return splitUsers(file, uacs.length); + } 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) + ) + ); + } await callees.reduce( (previous, spec) => previous.then(() => startRole(spec)), Promise.resolve() @@ -646,7 +690,7 @@ export function createRunBenchmark(deps: AgentDeps, state: RunState) { } await deps.log(`[${benchmarkId}] ${registerUsers} callees registered`); } - await startRole(uac); + await Promise.all(uacs.map((spec) => startRole(spec))); const startedMs = await sippStartTime( rawDir, uac.scenario, diff --git a/src/orchestrate/evaluateCapacityStep.ts b/src/orchestrate/evaluateCapacityStep.ts index cf9b983..74d94eb 100644 --- a/src/orchestrate/evaluateCapacityStep.ts +++ b/src/orchestrate/evaluateCapacityStep.ts @@ -7,7 +7,8 @@ import { measureCapacityStep, parseSamplerJsonl, parseUdpJsonl, - readSippResults + readSippResults, + uacProcessNames } from "../analyze/index.js"; import type { CapacityBenchmarkConfig, LivePoint } from "../schemas/index.js"; import { liveAnalyzeDeps } from "./liveAnalyzeDeps.js"; @@ -72,6 +73,7 @@ export async function evaluateCapacityStep(input: { ), udp: parseUdpJsonl(await read("udp-loadgen.jsonl"), "udp-loadgen.jsonl"), uutProcesses: [], + uacProcesses: uacProcessNames(benchmark.load.instances), unit: benchmark.load.unit }); const { p99LatencyMs, failedCallsPct } = benchmark.thresholds; diff --git a/src/orchestrate/sampler.ts b/src/orchestrate/sampler.ts index 145bb26..8964f5e 100644 --- a/src/orchestrate/sampler.ts +++ b/src/orchestrate/sampler.ts @@ -1,7 +1,13 @@ /** * Copyright (C) 2026 by Fonoster Inc (https://fonoster.com). MIT License. */ -import { SAMPLE_INTERVAL_S, SIPP_PROCESS, type SippRole } from "./constants.js"; +import { uacProcessName } from "../analyze/measureCapacityStep.js"; +import { + roleOf, + SAMPLE_INTERVAL_S, + SIPP_PROCESS, + type SippRole +} from "./constants.js"; import type { Host } from "./types.js"; /** @@ -45,18 +51,29 @@ 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 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. */ export function buildSamplerSpec( host: Host, processes: ReadonlyArray<{ name: string; match: string }>, - prefix = "" + prefix = "", + uacInstances = 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 }] + ); const entries = host === "uut" ? processes.map(({ name, match }) => [name, match]) - : (Object.keys(SIPP_PROCESS) as SippRole[]).map((role) => [ - SIPP_PROCESS[role], + : roles.map(({ name, role }) => [ + name, `sipp .*-cid_str ${prefix}${role}-` ]); return entries.map(([name, match]) => `${name}\t${match}\n`).join(""); diff --git a/src/schemas/perfkitConfig.schema.ts b/src/schemas/perfkitConfig.schema.ts index 3afc5b3..6350198 100644 --- a/src/schemas/perfkitConfig.schema.ts +++ b/src/schemas/perfkitConfig.schema.ts @@ -216,6 +216,17 @@ const loadFields = { /** The UUT's SIP port on the single server; unused with `uut.helm`. */ port: portSchema.default(5060), uac: uacSchema, + /** + * How many SIPp processes place the calls: the rate, the open-call limit + * and the `uac.users` file are split among them. SIPp is single-threaded, + * so one process tops out at about one core's worth of calls. + */ + instances: z + .number() + .int({ error: "expected a whole number of SIPp instances" }) + .min(1, { error: "expected 1 instance or more" }) + .max(64, { error: "expected 64 instances or fewer" }) + .default(1), hold: durationSchema.default("0s"), // What one SIPp "call" is, as a plural noun for the page: "calls", "messages". unit: z diff --git a/test/analyze/measureCapacityStep.unit.test.ts b/test/analyze/measureCapacityStep.unit.test.ts index db6d3fd..4797a52 100644 --- a/test/analyze/measureCapacityStep.unit.test.ts +++ b/test/analyze/measureCapacityStep.unit.test.ts @@ -28,7 +28,8 @@ function calleeAt(name: string, pct: number): string[] { async function step( spec: Parameters[0][number], - extraSampler: string[] = [] + extraSampler: string[] = [], + uacProcesses?: string[] ) { const files = capacityRaw([spec]); files["sampler.jsonl"] = [ @@ -51,6 +52,7 @@ async function step( ), udp: parseUdpJsonl(files["udp.jsonl"] as string, "udp.jsonl"), uutProcesses: ["edgeport"], + ...(uacProcesses ? { uacProcesses } : {}), unit: "calls" }); } @@ -83,6 +85,31 @@ describe("measureCapacityStep", () => { expect(below.invalidReason).to.equal(null); }); + it("should judge each uac instance on its own core, and name the busy one", async () => { + // Arrange: three instances, the third at 86% of its core, the rest idle + const uacProcesses = ["sipp-uac", "sipp-uac-2", "sipp-uac-3"]; + + // Act + const busy = await step( + { offeredCps: 40, attempted: 2400 }, + [...calleeAt("sipp-uac-2", 20), ...calleeAt("sipp-uac-3", 86)], + uacProcesses + ); + const fine = await step( + { offeredCps: 40, attempted: 2400 }, + uacProcesses.slice(1).flatMap((name) => calleeAt(name, 70)), + uacProcesses + ); + + // Assert: the busiest instance decides, not their sum (70 + 70 + 10 > 80) + expect(busy.invalidReason).to.equal( + "load generator saturated at 40 calls/s (SIPp instance sipp-uac-3 at 86% of one core)" + ); + expect(busy.step.sippCpuPct).to.equal(86); + expect(fine.invalidReason).to.equal(null); + expect(fine.step.sippCpuPct).to.equal(70); + }); + it("should measure a valid step", async () => { // Act const measured = await step({ diff --git a/test/analyze/windows.unit.test.ts b/test/analyze/windows.unit.test.ts index 8a09670..59b2879 100644 --- a/test/analyze/windows.unit.test.ts +++ b/test/analyze/windows.unit.test.ts @@ -103,4 +103,27 @@ describe("sumStatRows", () => { sumStatRows(rows, { fromMs: 15_000, toMs: 30_000 }).attempted ).to.equal(700); }); + + it("should add up each SIPp process's peak concurrent calls", () => { + // Arrange: two instances' rows, each peaking at its own time + const row = (pid: string, tMs: number, current: number) => ({ + tMs, + periodStartMs: tMs - 10_000, + counters: { "OutgoingCall(P)": 10, CurrentCall: current }, + pid + }); + const rows = [ + row("11", 20_000, 40), + row("11", 30_000, 50), + row("12", 20_100, 60), + row("12", 30_100, 30) + ]; + + // Act + const sums = sumStatRows(rows, { fromMs: 0, toMs: 60_000 }); + + // Assert + expect(sums.maxCurrentCalls).to.equal(110); + expect(sums.attempted).to.equal(40); + }); }); diff --git a/test/orchestrate/agent.unit.test.ts b/test/orchestrate/agent.unit.test.ts index 71d288d..18e92a4 100644 --- a/test/orchestrate/agent.unit.test.ts +++ b/test/orchestrate/agent.unit.test.ts @@ -326,6 +326,74 @@ describe("createAgent", () => { expect(progress).to.include("[one-soak] 200 callees registered"); }); + it("should start one SIPp per instance, each with its own share of the callers", async () => { + // Arrange + const dir = await mkdtemp(join(tmpdir(), "perfkit-instances-")); + await cp(dirname(ROUTR_CONFIG), dir, { recursive: true }); + const yaml = await readFile(ROUTR_CONFIG, "utf8"); + const edited = 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" + ); + expect(edited).to.not.equal(yaml); + await writeFile(join(dir, "perfkit.yaml"), edited); + const setup = await detached( + [ + "run", + "--config", + join(dir, "perfkit.yaml"), + "--target-version", + "2.31.5", + "--benchmarks", + "one-soak" + ], + soakRaw({ warmupS: 1800, afterS: 21600 }), + true + ); + await withCallees(setup); + const callers = Array.from({ length: 9 }, (_, i) => `caller${i};pw${i}`); + await writeFile( + join(setup.hosts.roots.loadgen, "project", "one-soak", "callers.csv"), + ["SEQUENTIAL", ...callers, ""].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); + const started = (name: string) => + commands.filter((c) => c.includes(name) && c.includes("run -d")); + expect(started("-sipp-uac ")).to.have.length(1); + 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"); + const raw = join(setup.hosts.roots.loadgen, "raw", "one-soak"); + const share = async (n: number) => + (await readFile(join(raw, `users-uac-${n}.csv`), "utf8")) + .trim() + .split("\n"); + expect(await share(1)).to.deep.equal([ + "SEQUENTIAL", + "caller0;pw0", + "caller3;pw3", + "caller6;pw6" + ]); + expect(await share(3)).to.deep.equal([ + "SEQUENTIAL", + "caller2;pw2", + "caller5;pw5", + "caller8;pw8" + ]); + // every instance is stopped, and the cleanup knows them all + expect( + commands.some( + (c) => c.includes("docker kill -s USR1") && c.includes("-sipp-uac3") + ) + ).to.equal(true); + }); + 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/buildSippRoles.unit.test.ts b/test/orchestrate/buildSippRoles.unit.test.ts index 76d7443..ca9bbd0 100644 --- a/test/orchestrate/buildSippRoles.unit.test.ts +++ b/test/orchestrate/buildSippRoles.unit.test.ts @@ -4,7 +4,9 @@ import { expect } from "chai"; import { buildSippRoles, - countUserRows + countUserRows, + splitUsers, + uacUsersFileOf } from "../../src/orchestrate/buildSippRoles.js"; import { capacityBenchmarkConfigSchema, @@ -164,6 +166,7 @@ describe("buildSippRoles", () => { expect(roles).to.deep.equal([ { role: "register", + instance: 1, scenario: "one-soak/uac_register.xml", args: [ "10.0.0.2:5060", @@ -199,6 +202,7 @@ describe("buildSippRoles", () => { }, { role: "uac", + instance: 1, scenario: "one-soak/uac_invite.xml", args: [ "10.0.0.2:5060", @@ -478,3 +482,96 @@ describe("countUserRows", () => { expect(countUserRows("")).to.equal(0); }); }); + +describe("buildSippRoles with several uac instances", () => { + const input = (benchmark: ReturnType) => ({ + benchmark, + runId: "RUN1", + uutIp: "10.0.0.2", + loadgenIp: "10.0.0.3", + registerUsers: 200 + }); + + it("should split a capacity ramp's rate and open calls among the instances", () => { + // Arrange + const benchmark = capacityWithCallees({ + instances: 4, + rate: { start: 100, step: 100, every: "60s", max: 400 } + }); + + // Act + const roles = buildSippRoles(input(benchmark)); + + // 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); + uacs.forEach(({ args, instance }) => { + const flags = flagsOf(args); + expect(flags).to.include("-r 25 -rp 1000 -rate_increase 25"); + expect(flags).to.include("-rate_max 100"); + // max 400 x (5 s hold + 20 s) = 10,000 calls, a quarter each + expect(flags).to.include("-l 2500"); + expect(flags).to.include(`-p ${5080 + instance - 1} `); + expect(flags).to.include( + `-cid_str ${instance === 1 ? "uac" : `uac${instance}`}-run1-%u-%p@%s` + ); + expect(flags).to.include(`-inf /raw/${uacUsersFileOf(instance)}`); + expect(flags).to.not.include("/project/one-soak/callers.csv"); + }); + }); + + it("should divide a soak's rate and give every instance a distinct port and Call-ID", () => { + // Arrange + const benchmark = soakBenchmarkConfigSchema.parse({ + ...soakConfig(), + load: { ...soakConfig().load, instances: 3, rate: { cps: 100 } } + }); + + // Act + const uacs = buildSippRoles(input(benchmark as never)).filter( + ({ role }) => role === "uac" + ); + + // Assert + expect(uacs).to.have.length(3); + uacs.forEach(({ args }) => expect(flagsOf(args)).to.include("-r 33.3333")); + const cids = uacs.map(({ args }) => args[args.indexOf("-cid_str") + 1]); + expect(new Set(cids).size).to.equal(3); + }); + + it("should keep the whole users file for a single instance", () => { + // Arrange + const benchmark = capacityWithCallees(); + + // Act + const uacs = buildSippRoles(input(benchmark)).filter( + ({ role }) => role === "uac" + ); + + // Assert + expect(uacs).to.have.length(1); + expect(flagsOf(uacs[0]!.args)).to.include( + "-inf /project/one-soak/callers.csv" + ); + }); +}); + +describe("splitUsers", () => { + it("should deal the rows out round-robin, each share with the mode line", () => { + // Act + const shares = splitUsers("SEQUENTIAL\na\nb\nc\nd\ne\n", 2); + + // Assert + expect(shares).to.deep.equal([ + "SEQUENTIAL\na\nc\ne\n", + "SEQUENTIAL\nb\nd\n" + ]); + }); + + it("should throw when there are fewer users than instances", () => { + expect(() => splitUsers("SEQUENTIAL\na\n", 2)).to.throw( + "the users file has 1 users, fewer than the 2 SIPp instances" + ); + }); +}); diff --git a/test/orchestrate/helpers.unit.test.ts b/test/orchestrate/helpers.unit.test.ts index bc3e223..ff29371 100644 --- a/test/orchestrate/helpers.unit.test.ts +++ b/test/orchestrate/helpers.unit.test.ts @@ -149,6 +149,15 @@ describe("orchestration helpers", () => { ); }); + it("should name each uac instance's SIPp process apart, in the cluster lane too", () => { + expect(buildSamplerSpec("loadgen", [], "", 3)).to.equal( + "sipp-uac\tsipp .*-cid_str uac-\nsipp-uac-2\tsipp .*-cid_str uac2-\nsipp-uac-3\tsipp .*-cid_str uac3-\nsipp-register\tsipp .*-cid_str register-\nsipp-uas\tsipp .*-cid_str uas-\n" + ); + expect(buildSamplerSpec("loadgen", [], "k8s-", 2)).to.include( + "sipp-uac-2\tsipp .*-cid_str k8s-uac2-\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/schemas/perfkitConfig.unit.test.ts b/test/schemas/perfkitConfig.unit.test.ts index a03c811..96d1659 100644 --- a/test/schemas/perfkitConfig.unit.test.ts +++ b/test/schemas/perfkitConfig.unit.test.ts @@ -110,6 +110,20 @@ describe("perfkitConfigSchema", () => { expect(config.benchmarks[0]?.load.uac.users).to.equal(undefined); }); + it("should default load.instances to 1 and accept 1 to 64", () => { + // Arrange + const raw = loadConfig(); + + // Act and Assert + expect(perfkitConfigSchema.parse(raw).benchmarks[0]?.load.instances).to.equal(1); + raw.benchmarks[0].load.instances = 8; + expect(perfkitConfigSchema.parse(raw).benchmarks[0]?.load.instances).to.equal(8); + [0, 65, 1.5].forEach((bad) => { + raw.benchmarks[0].load.instances = bad; + expect(() => perfkitConfigSchema.parse(raw)).to.throw(); + }); + }); + it("should default a capacity warm-up to 0s and accept one shorter than a step", () => { const raw = loadConfig(); expect(perfkitConfigSchema.parse(raw).benchmarks[0]?.warmup).to.equal( From 68b461cdac44469e5345acf857ad0c586d2c172a Mon Sep 17 00:00:00 2001 From: Pedro Sanders Date: Mon, 9 Nov 2026 06:16:14 -0400 Subject: [PATCH 2/3] feat: thresholds.confirmSteps ends a capacity ramp only after steps in a row break a rule A step that breaks p99LatencyMs or failedCallsPct while the next one passes is a spike: the ramp goes on, the step is flagged spike in the report and on the live page, and the limit stays the last step before a confirmed break. Default 2; 1 keeps the old behavior. A break on the last step and an unmeasurable (saturated load generator) step still end the ramp at once. Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_01WSt25UoKGj2mgDNQ4dHVrB --- docs/perfkit-yaml.md | 3 +- docs/report-v1.md | 2 +- schemas/live-v1.schema.json | 5 + schemas/perfkit-v1.schema.json | 6 + schemas/report-v1.schema.json | 6 + src/analyze/createAnalyzeCapacity.ts | 23 ++-- src/analyze/findRampEnd.ts | 56 ++++++++++ src/live/buildLiveCapacity.ts | 10 +- src/orchestrate/capacityLimitOf.ts | 12 +- src/orchestrate/createRunBenchmark.ts | 11 +- src/orchestrate/evaluateCapacityStep.ts | 41 +++++-- src/render/charts/renderLatencyChart.ts | 4 +- src/render/live/renderLiveCapacity.ts | 14 ++- src/schemas/live.schema.ts | 3 +- src/schemas/perfkitConfig.schema.ts | 4 +- src/schemas/report.schema.ts | 4 +- test/analyze/builders.ts | 4 +- .../createAnalyzeCapacity.unit.test.ts | 104 ++++++++++++++++++ test/analyze/findRampEnd.unit.test.ts | 64 +++++++++++ test/orchestrate/agent.unit.test.ts | 55 ++++++++- test/orchestrate/capacityLimitOf.unit.test.ts | 13 +++ 21 files changed, 402 insertions(+), 42 deletions(-) create mode 100644 src/analyze/findRampEnd.ts create mode 100644 test/analyze/findRampEnd.unit.test.ts diff --git a/docs/perfkit-yaml.md b/docs/perfkit-yaml.md index a733122..1127e33 100644 --- a/docs/perfkit-yaml.md +++ b/docs/perfkit-yaml.md @@ -173,10 +173,11 @@ Thresholds: | :--- | :--- | :--- | :--- | | `thresholds.p99LatencyMs` | capacity | ms | p99 time from INVITE to 200 OK. A step at or above it breaks the rule. | | `thresholds.failedCallsPct` | capacity, soak | % (0 to 100) | Failed calls. | +| `thresholds.confirmSteps` | capacity | integer 1 to 5, default 2 | How many steps in a row must break a rule to end the ramp. A step that breaks a rule while the next one passes is a spike: the ramp goes on, the step stays in the report flagged `spike`, and it is not the limit. `1` ends the ramp at the first break. A break on the last step counts as confirmed; a step the load generator could not measure ends the ramp at once. | | `thresholds.memoryGrowthMBPerHour` | soak | MB/h | Memory growth of each process after the warm-up. | | `thresholds.restarts` | soak | count | Process restarts allowed. | -`capacity` requires exactly `p99LatencyMs` and `failedCallsPct`. `soak` requires exactly +`capacity` requires `p99LatencyMs` and `failedCallsPct`, and takes `confirmSteps`. `soak` requires exactly `memoryGrowthMBPerHour`, `failedCallsPct` and `restarts`. A capacity uac gives up on any message it expects after 10 s (`-recv_timeout 10000`), and a call diff --git a/docs/report-v1.md b/docs/report-v1.md index 8cceb3a..5567176 100644 --- a/docs/report-v1.md +++ b/docs/report-v1.md @@ -115,7 +115,7 @@ Capacity `headline`: | `failedAtLimitPct`, `retransmissionsAtLimitPct` | number or null | Percentages at the limit. `null` when `limitCps` is. | Capacity `stoppedBy`: the step that ended the ramp, `{ offeredCps, rule, measured, limit }`, where -`rule` is `p99LatencyMs` or `failedCallsPct`. `null` when `limitIsLowerBound` is `true`. +`rule` is `p99LatencyMs` or `failedCallsPct`. A step may carry `spike: true` (optional, so older reports stay valid): it broke a rule but the next step passed, so it did not end the ramp. `null` when `limitIsLowerBound` is `true`. Capacity `series.steps`, one entry per step: diff --git a/schemas/live-v1.schema.json b/schemas/live-v1.schema.json index 1beb518..40f618a 100644 --- a/schemas/live-v1.schema.json +++ b/schemas/live-v1.schema.json @@ -231,6 +231,11 @@ "failedCallsPct": { "type": "number", "minimum": 0 + }, + "confirmSteps": { + "type": "integer", + "exclusiveMinimum": 0, + "maximum": 9007199254740991 } }, "required": [ diff --git a/schemas/perfkit-v1.schema.json b/schemas/perfkit-v1.schema.json index 929182e..46b86bb 100644 --- a/schemas/perfkit-v1.schema.json +++ b/schemas/perfkit-v1.schema.json @@ -518,6 +518,12 @@ "type": "number", "minimum": 0, "maximum": 100 + }, + "confirmSteps": { + "default": 2, + "type": "integer", + "minimum": 1, + "maximum": 5 } }, "required": [ diff --git a/schemas/report-v1.schema.json b/schemas/report-v1.schema.json index 05ba2a4..4bebb09 100644 --- a/schemas/report-v1.schema.json +++ b/schemas/report-v1.schema.json @@ -566,6 +566,9 @@ }, "valid": { "type": "boolean" + }, + "spike": { + "type": "boolean" } }, "required": [ @@ -866,6 +869,9 @@ }, "valid": { "type": "boolean" + }, + "spike": { + "type": "boolean" } }, "required": [ diff --git a/src/analyze/createAnalyzeCapacity.ts b/src/analyze/createAnalyzeCapacity.ts index a749bfa..1514eea 100644 --- a/src/analyze/createAnalyzeCapacity.ts +++ b/src/analyze/createAnalyzeCapacity.ts @@ -27,6 +27,7 @@ import { parseUdpJsonl } from "./parseUdpJsonl.js"; import { capacityMetaSchema } from "./rawResults.schema.js"; import { readRawFile } from "./readRawFile.js"; import { readSippResults } from "./readSippResults.js"; +import { findRampEnd } from "./findRampEnd.js"; import { roundTo } from "./roundTo.js"; import { toResources } from "./toResources.js"; import type { AnalyzeDeps } from "./types.js"; @@ -38,7 +39,8 @@ import type { AnalyzeDeps } from "./types.js"; * Steps are checked in order. A step whose measurement does not count (the * load generator saturated) makes the verdict inconclusive. Otherwise a step * passes when p99 < p99LatencyMs and failed % < failedCallsPct, both strict; - * the first step that breaks a rule ends the ramp. The limit is the offered + * the ramp ends where `thresholds.confirmSteps` steps in a row break a rule + * (a shorter run followed by a pass is a spike, flagged in the series). The limit is the offered * rate of the last step that passed, and the rules are reported at it, or at * the first step when none passed. When no step broke a rule the limit is a * lower bound. @@ -114,15 +116,22 @@ export function createAnalyzeCapacity(deps: AnalyzeDeps) { durationS: meta.durationS, tool: meta.tool }; - const series = { steps: measured.map(({ step }) => step) }; const resources = toResources(samples, uutProcesses); - // The ramp ends at the first step that is invalid or breaks a rule. - const endIndex = measured.findIndex( - (step) => - step.invalidReason !== null || - rulesOf(step).some((rule) => !rule.passed) + // The ramp ends at the first invalid step, or where confirmSteps steps in + // a row break a rule; a shorter run followed by a pass is a spike + const { endIndex, spikes } = findRampEnd( + measured.map((step) => ({ + valid: step.invalidReason === null, + broken: rulesOf(step).some((rule) => !rule.passed) + })), + benchmark.thresholds.confirmSteps ); + const series = { + steps: measured.map(({ step }, index) => + spikes.includes(index) ? { ...step, spike: true } : step + ) + }; const end = endIndex === -1 ? undefined : measured[endIndex]; if (end?.invalidReason) { return { diff --git a/src/analyze/findRampEnd.ts b/src/analyze/findRampEnd.ts new file mode 100644 index 0000000..49d4ad7 --- /dev/null +++ b/src/analyze/findRampEnd.ts @@ -0,0 +1,56 @@ +/** + * Copyright (C) 2026 by Fonoster Inc (https://fonoster.com). MIT License. + */ + +export interface RampStepState { + /** False when the step's measurement does not count (load generator gave out). */ + valid: boolean; + /** True when the step broke a rule (p99 or failed calls). */ + broken: boolean; +} + +export interface RampEnd { + /** Index of the step that ends the ramp, or -1 when none does. */ + endIndex: number; + /** Indexes of the steps that broke a rule but were not confirmed. */ + spikes: number[]; +} + +/** + * Where a capacity ramp ends. An invalid step ends it at once. A step that + * breaks a rule ends it only when `confirmSteps` steps in a row break one + * (the run starts at the first of them); a shorter run followed by a step + * that passes is a spike: the ramp goes on and the steps are listed in + * `spikes`. A run that reaches the last step counts as confirmed, and so does + * a run when the ramp stopped for another reason than a passing step. + * + * @param steps - The steps in order + * @param confirmSteps - How many consecutive steps must break a rule + */ +export function findRampEnd( + steps: readonly RampStepState[], + confirmSteps: number +): RampEnd { + const spikes: number[] = []; + const scan = (i: number): number => { + const step = steps[i]; + if (!step) return -1; + if (!step.valid) return i; + if (!step.broken) return scan(i + 1); + const runEnd = steps.findIndex( + (next, index) => index > i && !(next.valid && next.broken) + ); + const end = runEnd === -1 ? steps.length : runEnd; + const runLength = end - i; + const ranOut = end === steps.length; + const passedAfter = !ranOut && steps[end]?.valid === true; + if (runLength >= confirmSteps || ranOut) return i; + if (passedAfter) { + spikes.push(...Array.from({ length: runLength }, (_, k) => i + k)); + return scan(end); + } + // The run is cut by an invalid step, which ends the ramp + return end; + }; + return { endIndex: scan(0), spikes }; +} diff --git a/src/live/buildLiveCapacity.ts b/src/live/buildLiveCapacity.ts index 022a616..cbdf722 100644 --- a/src/live/buildLiveCapacity.ts +++ b/src/live/buildLiveCapacity.ts @@ -31,8 +31,14 @@ export function buildLiveCapacity(input: { const total = Math.floor((max - start) / step) + 1; const state = liveStateOf(input.selected, record); const points = record?.points ?? []; - const firstStop = points.findIndex((point) => !point.ok); - const passed = firstStop === -1 ? points : points.slice(0, firstStop); + // A step that broke a rule but did not stop the ramp is a spike: it neither + // ends the ramp nor counts as held + const firstStop = points.findIndex( + (point) => !point.ok && point.stopReason !== null + ); + const passed = ( + firstStop === -1 ? points : points.slice(0, firstStop) + ).filter((point) => point.ok); const heldCps = passed.at(-1)?.offeredCps ?? null; const loadStartedMs = record?.loadStartedMs ?? null; const index = diff --git a/src/orchestrate/capacityLimitOf.ts b/src/orchestrate/capacityLimitOf.ts index b334fd0..39049f2 100644 --- a/src/orchestrate/capacityLimitOf.ts +++ b/src/orchestrate/capacityLimitOf.ts @@ -22,7 +22,10 @@ export interface CapacityLimit { */ export function capacityLimitOf( result: BenchmarkResult | null, - thresholds: CapacityBenchmarkConfig["thresholds"] + thresholds: Pick< + CapacityBenchmarkConfig["thresholds"], + "p99LatencyMs" | "failedCallsPct" + > ): CapacityLimit | null { if (result === null || result.kind !== "capacity") return null; if (result.verdict === "passed") { @@ -39,9 +42,10 @@ export function capacityLimitOf( const measured = invalid === -1 ? steps : steps.slice(0, invalid); const passed = measured.every( (step) => - step.p99Ms < thresholds.p99LatencyMs && - (step.calls > 0 ? (step.failed / step.calls) * 100 : 0) < - thresholds.failedCallsPct + step.spike === true || + (step.p99Ms < thresholds.p99LatencyMs && + (step.calls > 0 ? (step.failed / step.calls) * 100 : 0) < + thresholds.failedCallsPct) ); const last = measured.at(-1); return passed && last ? { cps: last.offeredCps, lowerBound: true } : null; diff --git a/src/orchestrate/createRunBenchmark.ts b/src/orchestrate/createRunBenchmark.ts index 5b18933..e600747 100644 --- a/src/orchestrate/createRunBenchmark.ts +++ b/src/orchestrate/createRunBenchmark.ts @@ -785,7 +785,8 @@ export function createRunBenchmark(deps: AgentDeps, state: RunState) { const everyMs = durationToSeconds(every) * 1000; const total = Math.floor((max - start) / step) + 1; const runStep = async ( - index: number + index: number, + priorBreaks: number ): Promise<{ steps: number; aborted: string | null }> => { const ok = await waitUntil( startedMs + (index + 1) * everyMs + STEP_GRACE_MS @@ -796,16 +797,18 @@ export function createRunBenchmark(deps: AgentDeps, state: RunState) { rawDir, benchmark, index, - startMs: startedMs + startMs: startedMs, + priorBreaks, + lastStep: index + 1 >= total }); await deps.log(evaluation.line); if (evaluation.point) deps.live?.point(benchmark.id, evaluation.point); if (evaluation.stop || index + 1 >= total) { return { steps: index + 1, aborted: null }; } - return runStep(index + 1); + return runStep(index + 1, evaluation.broken ? priorBreaks + 1 : 0); }; - return runStep(0); + return runStep(0, 0); } async function collectUutFiles(rawDir: string) { diff --git a/src/orchestrate/evaluateCapacityStep.ts b/src/orchestrate/evaluateCapacityStep.ts index 74d94eb..14b445b 100644 --- a/src/orchestrate/evaluateCapacityStep.ts +++ b/src/orchestrate/evaluateCapacityStep.ts @@ -17,8 +17,10 @@ import type { AgentFs } from "./types.js"; export interface StepEvaluation { /** The progress line, like `[id] step 50 calls/s p99 4.1 ms ... ok`. */ line: string; - /** True when the ramp ends here: a rule broken or the step invalid. */ + /** True when the ramp ends here: the step invalid, or confirmSteps in a row broke a rule. */ stop: boolean; + /** True when the step broke a rule (it ends the ramp only once confirmed). */ + broken: boolean; /** The step's numbers for live.json; null when SIPp wrote nothing. */ point: LivePoint | null; } @@ -35,6 +37,10 @@ export async function evaluateCapacityStep(input: { benchmark: CapacityBenchmarkConfig; index: number; startMs: number; + /** How many steps just before this one broke a rule, in a row; 0 when none. */ + priorBreaks?: number; + /** True for the last step of the ramp: a break there has no next step to wait for. */ + lastStep?: boolean; }): Promise { const { benchmark, index, rawDir } = input; const { start, step, every } = benchmark.load.rate; @@ -53,6 +59,7 @@ export async function evaluateCapacityStep(input: { return { line: `${prefix} SIPp wrote no results; stopping`, stop: true, + broken: false, point: null }; } @@ -79,18 +86,30 @@ export async function evaluateCapacityStep(input: { const { p99LatencyMs, failedCallsPct } = benchmark.thresholds; const p99 = measured.step.p99Ms; const numbers = `${prefix} p99 ${p99.toFixed(1)} ms failed ${measured.failedPct.toFixed(2)}%`; - const outcome = - measured.invalidReason !== null - ? `invalid, ${measured.invalidReason}; stopping` - : p99 >= p99LatencyMs - ? "over the p99 limit; stopping" - : measured.failedPct >= failedCallsPct - ? "over the failed limit; stopping" - : "ok"; - const stop = outcome !== "ok"; + const invalid = measured.invalidReason !== null; + const brokenRule = + p99 >= p99LatencyMs + ? "over the p99 limit" + : measured.failedPct >= failedCallsPct + ? "over the failed limit" + : null; + const broken = !invalid && brokenRule !== null; + const confirmed = + broken && + (input.lastStep === true || + (input.priorBreaks ?? 0) + 1 >= benchmark.thresholds.confirmSteps); + const stop = invalid || confirmed; + const outcome = invalid + ? `invalid, ${measured.invalidReason}; stopping` + : broken + ? confirmed + ? `${brokenRule}; stopping` + : `${brokenRule}; a spike unless the next step breaks too` + : "ok"; return { line: `${numbers} ${outcome}`, stop, + broken, point: { offeredCps, achievedCps: measured.step.achievedCps, @@ -100,7 +119,7 @@ export async function evaluateCapacityStep(input: { p95Ms: measured.step.p95Ms, p99Ms: p99, failedPct: measured.failedPct, - ok: !stop, + ok: !invalid && !broken, stopReason: stop ? outcome.replace(/; stopping$/, "") : null } }; diff --git a/src/render/charts/renderLatencyChart.ts b/src/render/charts/renderLatencyChart.ts index 1912994..2c2fbbe 100644 --- a/src/render/charts/renderLatencyChart.ts +++ b/src/render/charts/renderLatencyChart.ts @@ -128,14 +128,14 @@ ${live ? liveRegion(live, steps, sx) : ""}${overLimit}${threshold}${lines}${limi const rows = steps .map( (s) => - `${formatNumber(s.offeredCps)}${formatNumber(s.achievedCps)}${formatNumber(s.p50Ms)}${formatNumber(s.p95Ms)}${formatNumber(s.p99Ms)}${formatPct(s.calls === 0 ? 0 : (s.failed / s.calls) * 100)}${s.valid ? "yes" : "no"}` + `${formatNumber(s.offeredCps)}${formatNumber(s.achievedCps)}${formatNumber(s.p50Ms)}${formatNumber(s.p95Ms)}${formatNumber(s.p99Ms)}${formatPct(s.calls === 0 ? 0 : (s.failed / s.calls) * 100)}${s.spike ? "spike" : s.valid ? "yes" : "no"}` ) .join(""); return `

Latency as load rises

request to 200 OK, by offered rate

-
${input.partial ? 'Partial' : ""}
    ${legend}
+
${input.partial ? 'Partial' : ""}${steps.some((s) => s.spike) ? `${steps.filter((s) => s.spike).length} ${steps.filter((s) => s.spike).length === 1 ? "spike" : "spikes"}` : ""}
    ${legend}
${svg}
diff --git a/src/render/live/renderLiveCapacity.ts b/src/render/live/renderLiveCapacity.ts index ff1534f..fe9bed7 100644 --- a/src/render/live/renderLiveCapacity.ts +++ b/src/render/live/renderLiveCapacity.ts @@ -46,7 +46,8 @@ export function renderLiveCapacity( uutCpuPct: 0, loadgenUdpDrops: 0, uutUdpDrops: 0, - valid: point.ok || point.stopReason === null + valid: point.ok || point.stopReason === null, + ...(!point.ok && point.stopReason === null ? { spike: true } : {}) })); const done = benchmark.state === "done"; const held = benchmark.heldCps; @@ -75,6 +76,11 @@ export function renderLiveCapacity( }); const passed = benchmark.points.filter((p) => p.ok).length; + const spikes = steps.filter((s) => s.spike).length; + const spikeNote = + spikes === 0 + ? "" + : ` ${spikes} ${spikes === 1 ? "step broke a rule" : "steps broke a rule"} but the next one passed: a spike, not the limit.`; const value = done ? held === null ? "No step held" @@ -84,9 +90,9 @@ export function renderLiveCapacity( : `Held ${rateOf(held)} so far`; const sentence = done ? benchmark.result?.stopReason - ? `The ramp stopped at ${rateOf(benchmark.points.find((p) => !p.ok)?.offeredCps ?? 0)}: ${benchmark.result.stopReason}. The verdict comes from the full analysis when the run is collected.` - : "Every step passed up to the highest rate offered. The verdict comes from the full analysis when the run is collected." - : `${passed === 0 ? "No step has finished yet." : `${passed} ${passed === 1 ? "step" : "steps"} passed both rules.`} The limit is the last step that passes before one fails; it is not known until the ramp stops.`; + ? `The ramp stopped at ${rateOf(benchmark.points.find((p) => !p.ok && p.stopReason !== null)?.offeredCps ?? 0)}: ${benchmark.result.stopReason}. The verdict comes from the full analysis when the run is collected.` + : `Every step passed up to the highest rate offered.${spikeNote} The verdict comes from the full analysis when the run is collected.` + : `${passed === 0 ? "No step has finished yet." : `${passed} ${passed === 1 ? "step" : "steps"} passed both rules.`}${spikeNote} The limit is the last step that passes before the ramp stops; it is not known until then.`; const last = lastPublishedLimit(ctx.runs, benchmark.id, ctx.live.runId); const baseline = last ? `Last published result: ${rateOf(last.limitCps)} on v${last.version}.${ diff --git a/src/schemas/live.schema.ts b/src/schemas/live.schema.ts index be84deb..1e8532a 100644 --- a/src/schemas/live.schema.ts +++ b/src/schemas/live.schema.ts @@ -78,7 +78,8 @@ export const liveCapacitySchema = z.strictObject({ }), thresholds: z.strictObject({ p99LatencyMs: positiveNumber, - failedCallsPct: nonNegativeNumber + failedCallsPct: nonNegativeNumber, + confirmSteps: positiveNumber.int().optional() }), /** The step SIPp is offering now, while running; else null. */ current: z diff --git a/src/schemas/perfkitConfig.schema.ts b/src/schemas/perfkitConfig.schema.ts index 6350198..64ec19a 100644 --- a/src/schemas/perfkitConfig.schema.ts +++ b/src/schemas/perfkitConfig.schema.ts @@ -292,7 +292,9 @@ const percentSchema = nonNegativeNumber.max(100, { const capacityThresholdsSchema = z.strictObject({ p99LatencyMs: positiveNumber, - failedCallsPct: percentSchema + failedCallsPct: percentSchema, + /** How many steps in a row must break a rule to end the ramp; 1 ends it at the first. */ + confirmSteps: z.number().int().min(1).max(5).default(2) }); const soakThresholdsSchema = z.strictObject({ diff --git a/src/schemas/report.schema.ts b/src/schemas/report.schema.ts index 504c303..af02c4b 100644 --- a/src/schemas/report.schema.ts +++ b/src/schemas/report.schema.ts @@ -189,7 +189,9 @@ export const stepSchema = z.strictObject({ uutCpuPct: nonNegativeNumber, loadgenUdpDrops: count, uutUdpDrops: count, - valid: z.boolean() + valid: z.boolean(), + /** True when the step broke a rule but the next one passed: a spike, not the limit. */ + spike: z.boolean().optional() }); export const seriesSchema = z.strictObject({ diff --git a/test/analyze/builders.ts b/test/analyze/builders.ts index 1a7d915..eb39603 100644 --- a/test/analyze/builders.ts +++ b/test/analyze/builders.ts @@ -60,6 +60,7 @@ export function capacityConfig( overrides: { p99LatencyMs?: number; failedCallsPct?: number; + confirmSteps?: number; max?: number; } = {}, extra: Record = {} @@ -85,7 +86,8 @@ export function capacityConfig( }, thresholds: { p99LatencyMs: overrides.p99LatencyMs ?? 50, - failedCallsPct: overrides.failedCallsPct ?? 0.1 + failedCallsPct: overrides.failedCallsPct ?? 0.1, + confirmSteps: overrides.confirmSteps ?? 1 }, ...extra }); diff --git a/test/analyze/createAnalyzeCapacity.unit.test.ts b/test/analyze/createAnalyzeCapacity.unit.test.ts index c270b54..98268f6 100644 --- a/test/analyze/createAnalyzeCapacity.unit.test.ts +++ b/test/analyze/createAnalyzeCapacity.unit.test.ts @@ -475,4 +475,108 @@ describe("createAnalyzeCapacity", () => { expect(error.message).to.include("no SIPp -trace_stat file"); }); }); + describe("spike tolerance (confirmSteps)", () => { + const slow = latenciesWithP99(80); + + it("should treat a rule break followed by a pass as a spike and go on", async () => { + // Act + const result = measured( + await analyze( + [ + { offeredCps: 50 }, + { offeredCps: 100, latencies: slow }, + { offeredCps: 150 }, + { offeredCps: 200 } + ], + capacityConfig({ confirmSteps: 2 }) + ) + ); + + // Assert + expect(result.verdict).to.equal("passed"); + expect(result.headline.limitCps).to.equal(200); + expect(result.headline.limitIsLowerBound).to.equal(true); + expect(result.stoppedBy).to.equal(null); + expect(result.series.steps.map((step) => step.spike)).to.deep.equal([ + undefined, + true, + undefined, + undefined + ]); + }); + + it("should end the ramp at the first of two steps in a row that break a rule", async () => { + // Act + const result = measured( + await analyze( + [ + { offeredCps: 50 }, + { offeredCps: 100 }, + { offeredCps: 150, latencies: slow }, + { offeredCps: 200, latencies: slow } + ], + capacityConfig({ confirmSteps: 2 }) + ) + ); + + // Assert + expect(result.headline.limitCps).to.equal(100); + expect(result.headline.limitIsLowerBound).to.equal(false); + expect(result.stoppedBy).to.include({ + offeredCps: 150, + rule: "p99LatencyMs" + }); + expect(result.series.steps.some((step) => step.spike)).to.equal(false); + }); + + it("should count a break on the last step as confirmed", async () => { + // Act + const result = measured( + await analyze( + [ + { offeredCps: 50 }, + { offeredCps: 100 }, + { offeredCps: 150, latencies: slow } + ], + capacityConfig({ confirmSteps: 2 }) + ) + ); + + // Assert + expect(result.headline.limitCps).to.equal(100); + expect(result.stoppedBy).to.include({ offeredCps: 150 }); + }); + + it("should end the ramp at the old first break with confirmSteps 1", async () => { + // Act + const result = measured( + await analyze( + [ + { offeredCps: 50 }, + { offeredCps: 100, latencies: slow }, + { offeredCps: 150 } + ], + capacityConfig({ confirmSteps: 1 }) + ) + ); + + // Assert + expect(result.headline.limitCps).to.equal(50); + }); + + it("should stay inconclusive when the load generator gives out after a spike", async () => { + // Act + const result = await analyze( + [ + { offeredCps: 50 }, + { offeredCps: 100, latencies: slow }, + { offeredCps: 150, sippCpuPct: 93 } + ], + capacityConfig({ confirmSteps: 2 }) + ); + + // Assert + expect(result.verdict).to.equal("inconclusive"); + }); + }); }); diff --git a/test/analyze/findRampEnd.unit.test.ts b/test/analyze/findRampEnd.unit.test.ts new file mode 100644 index 0000000..d496b4a --- /dev/null +++ b/test/analyze/findRampEnd.unit.test.ts @@ -0,0 +1,64 @@ +/** + * Copyright (C) 2026 by Fonoster Inc (https://fonoster.com). MIT License. + */ +import { expect } from "chai"; +import { findRampEnd } from "../../src/analyze/findRampEnd.js"; + +const ok = { valid: true, broken: false }; +const bad = { valid: true, broken: true }; +const invalid = { valid: false, broken: false }; + +describe("findRampEnd", () => { + it("should not end a ramp where every step passes", () => { + expect(findRampEnd([ok, ok, ok], 2)).to.deep.equal({ + endIndex: -1, + spikes: [] + }); + }); + + it("should end at the first break with confirmSteps 1", () => { + expect(findRampEnd([ok, bad, ok], 1)).to.deep.equal({ + endIndex: 1, + spikes: [] + }); + }); + + it("should list a break that the next step passes as a spike", () => { + expect(findRampEnd([ok, bad, ok, ok], 2)).to.deep.equal({ + endIndex: -1, + spikes: [1] + }); + }); + + it("should end at the start of a confirmed run", () => { + expect(findRampEnd([ok, bad, ok, bad, bad], 2)).to.deep.equal({ + endIndex: 3, + spikes: [1] + }); + }); + + it("should list every step of a run shorter than confirmSteps that a pass ends", () => { + expect(findRampEnd([ok, bad, bad, ok], 3)).to.deep.equal({ + endIndex: -1, + spikes: [1, 2] + }); + }); + + it("should confirm a run that reaches the last step", () => { + expect(findRampEnd([ok, ok, bad], 3)).to.deep.equal({ + endIndex: 2, + spikes: [] + }); + }); + + it("should end at an invalid step at once", () => { + expect(findRampEnd([ok, invalid, ok], 2).endIndex).to.equal(1); + }); + + it("should end at the invalid step that follows an unconfirmed break", () => { + expect(findRampEnd([ok, bad, invalid], 2)).to.deep.equal({ + endIndex: 2, + spikes: [] + }); + }); +}); diff --git a/test/orchestrate/agent.unit.test.ts b/test/orchestrate/agent.unit.test.ts index 18e92a4..deb73aa 100644 --- a/test/orchestrate/agent.unit.test.ts +++ b/test/orchestrate/agent.unit.test.ts @@ -86,7 +86,7 @@ async function routrWithCapacityCallees(): Promise { uas: { scenario: one-soak/uas_answer.xml } uac: { scenario: edgeport-echo/uac_message.xml } rate: { start: 50, step: 50, every: 60s, max: 200 } - thresholds: { p99LatencyMs: 150, failedCallsPct: 0.1 } + thresholds: { p99LatencyMs: 150, failedCallsPct: 0.1, confirmSteps: 1 } ` ); return join(dir, "perfkit.yaml"); @@ -107,6 +107,24 @@ function capacityWithRegistrations(registered: number) { const agentCommands = (setup: Awaited>) => setup.hosts.commands.filter(({ by }) => by === "agent"); +/** A copy of the minimal example whose capacity thresholds carry `extra`. */ +async function minimalWithThresholds(extra: string): Promise { + const dir = await mkdtemp(join(tmpdir(), "perfkit-thresholds-")); + await cp(dirname(MINIMAL_CONFIG), dir, { recursive: true }); + const config = join(dir, "perfkit.yaml"); + const yaml = await readFile(config, "utf8"); + await writeFile( + config, + yaml + .replace( + "thresholds: { p99LatencyMs: 50, failedCallsPct: 0.1 }", + `thresholds: { p99LatencyMs: 50, failedCallsPct: 0.1, ${extra} }` + ) + .replace("max: 200", "max: 300") + ); + return ["run", "--config", config, "--target-version", "3.7.3"]; +} + describe("createAgent", () => { it("should stop SIPp right after the first step that breaks a rule and start no next step", async () => { // Arrange @@ -116,7 +134,10 @@ describe("createAgent", () => { { offeredCps: 150 }, { offeredCps: 200 } ]; - const setup = await detached(MINIMAL, capacityRaw(steps)); + const setup = await detached( + await minimalWithThresholds("confirmSteps: 1"), + capacityRaw(steps) + ); const progress = join(setup.hosts.roots.loadgen, "progress.log"); // Act @@ -146,6 +167,36 @@ describe("createAgent", () => { ); }); + it("should go on after a spike and stop when the next step breaks a rule too", async () => { + // Arrange: step 100 is slow but step 150 passes (a spike); 200 and 250 are slow + const steps = [ + { offeredCps: 50 }, + { offeredCps: 100, latencies: latenciesWithP99(80) }, + { offeredCps: 150 }, + { offeredCps: 200, latencies: latenciesWithP99(80) }, + { offeredCps: 250, latencies: latenciesWithP99(80) }, + { offeredCps: 300 } + ]; + const setup = await detached( + await minimalWithThresholds("confirmSteps: 2"), + capacityRaw(steps) + ); + + // Act + await setup.agent.run(setup.runId()); + + // Assert + const progress = join(setup.hosts.roots.loadgen, "progress.log"); + const lines = (await readFile(progress, "utf8")).trim().split("\n"); + expect(lines).to.deep.equal([ + "[sipp-uas] step 50 messages/s p99 5.0 ms failed 0.00% ok", + "[sipp-uas] step 100 messages/s p99 80.0 ms failed 0.00% over the p99 limit; a spike unless the next step breaks too", + "[sipp-uas] step 150 messages/s p99 5.0 ms failed 0.00% ok", + "[sipp-uas] step 200 messages/s p99 80.0 ms failed 0.00% over the p99 limit; a spike unless the next step breaks too", + "[sipp-uas] step 250 messages/s p99 80.0 ms failed 0.00% over the p99 limit; stopping" + ]); + }); + it("should leave a capacity warm-up out of the first step, live and in the final verdict", async () => { // Arrange: the first 30 of step 1's 100 messages (sent in its first 18 s) // are slow, as with a cold JIT; a 20 s warm-up leaves them out. diff --git a/test/orchestrate/capacityLimitOf.unit.test.ts b/test/orchestrate/capacityLimitOf.unit.test.ts index 4ad7456..6a83a1a 100644 --- a/test/orchestrate/capacityLimitOf.unit.test.ts +++ b/test/orchestrate/capacityLimitOf.unit.test.ts @@ -66,6 +66,19 @@ describe("capacityLimitOf", () => { }); }); + it("should not count a spike step as a rule break before the invalid one", () => { + const result = inconclusive([ + step(100), + step(200, { p99Ms: 500, spike: true }), + step(300), + step(400, { valid: false, achievedCps: 300 }) + ]); + expect(capacityLimitOf(result, THRESHOLDS)).to.deep.equal({ + cps: 300, + lowerBound: true + }); + }); + it("should give no limit when a step before the invalid one broke a rule, or the first step is invalid", () => { expect( capacityLimitOf( From c41188a30e1555d10521bbfa96c132fbb2f0f3b5 Mon Sep 17 00:00:00 2001 From: Pedro Sanders Date: Mon, 9 Nov 2026 09:14:20 -0400 Subject: [PATCH 3/3] feat: infra.loadgen.count adds load generator droplets that run uac SIPp processes load.instances stays the total and is spread round robin over the droplets; each worker's SIPp files are pulled back to the first droplet over SSH, with their pids renamed so two machines never collide. Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_01WSt25UoKGj2mgDNQ4dHVrB --- README.md | 2 +- docs/design-loadgen-count.md | 5 +- docs/perfkit-yaml.md | 14 ++- schemas/perfkit-v1.schema.json | 6 + src/cli/createStartExecutor.ts | 15 ++- src/cli/formatPlan.ts | 2 +- src/infra/createPlanRun.ts | 9 ++ src/infra/createProvision.ts | 19 ++- src/infra/createSweep.ts | 12 +- src/infra/infra.schema.ts | 6 +- src/infra/types.ts | 4 +- src/orchestrate/buildSippRoles.ts | 6 +- src/orchestrate/createAgentShell.ts | 27 ++++- src/orchestrate/createRunBenchmark.ts | 120 +++++++++++++++++-- src/orchestrate/createSshRemote.ts | 8 +- src/orchestrate/createStartRun.ts | 23 +++- src/orchestrate/runState.schema.ts | 4 + src/orchestrate/types.ts | 11 +- src/orchestrate/workerFiles.ts | 27 +++++ src/schemas/perfkitConfig.schema.ts | 16 ++- test/infra/createPlanRun.unit.test.ts | 27 +++++ test/infra/createProvision.unit.test.ts | 37 ++++++ test/infra/createSweep.unit.test.ts | 31 +++++ test/infra/fakes.ts | 8 +- test/orchestrate/buildSippRoles.unit.test.ts | 26 ++++ test/orchestrate/workerFiles.unit.test.ts | 63 ++++++++++ test/schemas/perfkitConfig.unit.test.ts | 14 +++ 27 files changed, 501 insertions(+), 41 deletions(-) create mode 100644 src/orchestrate/workerFiles.ts create mode 100644 test/orchestrate/workerFiles.unit.test.ts diff --git a/README.md b/README.md index 1108051..7d6fa1c 100644 --- a/README.md +++ b/README.md @@ -362,7 +362,7 @@ perfkit sweep [--config ] [--run ] [--older-than -uut` or `-loadgen`, +schedule. It deletes droplets tagged `perfkit` and named `perfkit--uut`, `-loadgen` or `-worker-`, and Kubernetes clusters tagged `perfkit` and named `perfkit-` (with their load balancers and volumes), that are older than `--older-than`, and then the firewall, VPC and SSH key named `perfkit-` of those runs. It touches nothing else: not an untagged droplet, and never the diff --git a/docs/design-loadgen-count.md b/docs/design-loadgen-count.md index 202745e..e99e915 100644 --- a/docs/design-loadgen-count.md +++ b/docs/design-loadgen-count.md @@ -1,6 +1,9 @@ # Design: several load generator droplets (`infra.loadgen.count`) -Status: not implemented. `load.instances` (SIPp processes on one load generator) ships first. +Status: implemented in a simpler shape than below: `load.instances` stays the total and is +spread round robin over the droplets; workers run uac only and their SIPp files are pulled +back to the first droplet over SSH (their pids are renamed so two machines never collide). +Not done: sampling the workers, and `sipp-uac@` per-process names. ## Why diff --git a/docs/perfkit-yaml.md b/docs/perfkit-yaml.md index 1127e33..6c17392 100644 --- a/docs/perfkit-yaml.md +++ b/docs/perfkit-yaml.md @@ -45,6 +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.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. | @@ -316,6 +317,13 @@ benchmarks: One c-16 load generator with `load.instances: 12` places about 12 times what one SIPp does. Beyond one machine the limits are network and source ports: over TCP each call holds a connection (`ip_local_port_range` is 1024-65535 with `tcp_tw_reuse`), so about 60,000 -connections per load generator address, which is `cps × (hold + TIME_WAIT)`. Several load -generator droplets (`infra.loadgen.count`) are the next step and are not implemented yet; see -the design in `docs/design-loadgen-count.md`. +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. +The design is in `docs/design-loadgen-count.md`. diff --git a/schemas/perfkit-v1.schema.json b/schemas/perfkit-v1.schema.json index 46b86bb..170f745 100644 --- a/schemas/perfkit-v1.schema.json +++ b/schemas/perfkit-v1.schema.json @@ -96,6 +96,12 @@ "size": { "type": "string", "minLength": 1 + }, + "count": { + "default": 1, + "type": "integer", + "minimum": 1, + "maximum": 8 } }, "required": [ diff --git a/src/cli/createStartExecutor.ts b/src/cli/createStartExecutor.ts index 6e8e141..6551474 100644 --- a/src/cli/createStartExecutor.ts +++ b/src/cli/createStartExecutor.ts @@ -41,7 +41,16 @@ export function createStartExecutor(deps: CliDeps): BenchmarkExecutor { remote: createSshRemote({ ssh: infra.ssh, privateKey: run.identity.privateKey, - publicIps: { uut: run.uut.publicIp, loadgen: run.loadgen.publicIp } + publicIps: { + uut: run.uut.publicIp, + loadgen: run.loadgen.publicIp, + ...Object.fromEntries( + run.workers.map((worker, index) => [ + `worker-${index + 1}`, + worker.publicIp + ]) + ) + } }), packageRoot: infra.packageRoot })({ @@ -58,6 +67,10 @@ export function createStartExecutor(deps: CliDeps): BenchmarkExecutor { hosts: { uutIp: run.uut.privateIp, loadgenIp: run.loadgen.privateIp, + workers: run.workers.map(({ publicIp, privateIp }) => ({ + publicIp, + privateIp + })), vpcCidr: run.vpc.ipRange }, projectDir: dirname(options.configPath), diff --git a/src/cli/formatPlan.ts b/src/cli/formatPlan.ts index e60cf6b..7868a5f 100644 --- a/src/cli/formatPlan.ts +++ b/src/cli/formatPlan.ts @@ -61,7 +61,7 @@ export function formatPlan( ]), ...field("uut", [`${plan.uut.size.padEnd(sizeWidth)}${plan.uut.image}`]), ...field("loadgen", [ - `${plan.loadgen.size.padEnd(sizeWidth)}${plan.loadgen.image}` + `${plan.loadgen.size.padEnd(sizeWidth)}${plan.loadgen.image}${plan.workers.length > 0 ? `, ${plan.workers.length + 1} of them` : ""}` ]), ...(plan.cluster ? field("cluster", [ diff --git a/src/infra/createPlanRun.ts b/src/infra/createPlanRun.ts index e6e890d..f0a50b1 100644 --- a/src/infra/createPlanRun.ts +++ b/src/infra/createPlanRun.ts @@ -75,6 +75,15 @@ export function createPlanRun(deps: { size: config.infra.loadgen.size, image: DROPLET_IMAGE }, + workers: Array.from( + { length: config.infra.loadgen.count - 1 }, + (_, index) => ({ + role: "worker" as const, + name: `perfkit-${runId}-worker-${index + 1}`, + size: config.infra.loadgen.size, + image: DROPLET_IMAGE + }) + ), sshSource: `${ip}/32`, images, benchmarks: selected, diff --git a/src/infra/createProvision.ts b/src/infra/createProvision.ts index 7b34a0c..a52f1ce 100644 --- a/src/infra/createProvision.ts +++ b/src/infra/createProvision.ts @@ -156,6 +156,12 @@ export function createProvision(deps: ProvisionDeps) { }; await createDroplet(plan.uut, options.authorizedKeys); await createDroplet(plan.loadgen, []); + // The other load generators take commands from the agent over SSH, like the UUT + await plan.workers.reduce( + (previous, worker) => + previous.then(() => createDroplet(worker, options.authorizedKeys)), + Promise.resolve() + ); const clusterPlan = plan.cluster; if (clusterPlan) { @@ -307,11 +313,15 @@ export function createProvision(deps: ProvisionDeps) { const droplets = async () => { const uut = await waitActive(0); const loadgen = await waitActive(1); - await waitCloudInit(uut); - await waitCloudInit(loadgen); - return { uut, loadgen }; + const workers = await Promise.all( + plan.workers.map((_, index) => waitActive(2 + index)) + ); + await Promise.all( + [uut, loadgen, ...workers].map((droplet) => waitCloudInit(droplet)) + ); + return { uut, loadgen, workers }; }; - const [{ uut, loadgen }, provisionedCluster] = await Promise.all([ + const [{ uut, loadgen, workers }, provisionedCluster] = await Promise.all([ droplets(), created.cluster ? waitCluster() : Promise.resolve(undefined) ]); @@ -324,6 +334,7 @@ export function createProvision(deps: ProvisionDeps) { firewall: created.firewall, uut, loadgen, + workers, identity, ...(provisionedCluster ? { provisionedCluster } : {}) }; diff --git a/src/infra/createSweep.ts b/src/infra/createSweep.ts index ec0ca71..ccb0063 100644 --- a/src/infra/createSweep.ts +++ b/src/infra/createSweep.ts @@ -20,7 +20,9 @@ import type { const RUN_ID = String.raw`(\d{8}T\d{6}Z-[0-9a-f]{7})`; const RESOURCE_NAME = new RegExp(`^perfkit-${RUN_ID}$`); -const DROPLET_NAME = new RegExp(`^perfkit-${RUN_ID}-(uut|loadgen)$`); +const DROPLET_NAME = new RegExp( + `^perfkit-${RUN_ID}-(uut|loadgen|worker-\\d+)$` +); const RUN_TAG = new RegExp(`^perfkit-run-${RUN_ID}$`); const CLUSTER_NAME = /^perfkit-(\d{8})t(\d{6})z-([0-9a-f]{7})$/; @@ -43,7 +45,7 @@ export interface SweepResult { /** * Creates `sweep({ olderThan, runId, dryRun })`, the backstop for a run whose * teardown never happened. It deletes droplets tagged `perfkit` and named - * `perfkit--uut|loadgen` that are older than `olderThan`, or every + * `perfkit--uut|loadgen|worker-` that are older than `olderThan`, or every * one of run `runId` at any age. Once all of a run's droplets are selected * (or it has none left and its run id is older than `olderThan`), it also * deletes that run's firewall, VPC and SSH key, all named `perfkit-`. @@ -134,7 +136,11 @@ export function createSweep(deps: { droplets: selected.map(({ droplet }) => ({ id: droplet.id, name: droplet.name, - role: droplet.name.endsWith("-uut") ? "uut" : "loadgen", + role: droplet.name.endsWith("-uut") + ? "uut" + : /-worker-\d+$/.test(droplet.name) + ? "worker" + : "loadgen", publicIp: null, privateIp: null })), diff --git a/src/infra/infra.schema.ts b/src/infra/infra.schema.ts index 935327f..f688837 100644 --- a/src/infra/infra.schema.ts +++ b/src/infra/infra.schema.ts @@ -81,7 +81,7 @@ export const planRunInputSchema = z }); const dropletPlanSchema = z.strictObject({ - role: z.enum(["uut", "loadgen"]), + role: z.enum(["uut", "loadgen", "worker"]), name: z.string().regex(/^perfkit-[A-Za-z0-9.-]+$/, { error: "expected a droplet name starting with perfkit-" }), @@ -123,6 +123,8 @@ export const planSchema = z.strictObject({ tags: z.array(nonEmptyString).min(1, { error: "expected at least one tag" }), uut: dropletPlanSchema, loadgen: dropletPlanSchema, + /** More load generators (`infra.loadgen.count` - 1): SIPp uac processes only. */ + workers: z.array(dropletPlanSchema).default([]), sshSource: ipv4Cidr32Schema, images: z.array(resolvedImageSchema).min(1, { error: "expected at least one image" @@ -156,7 +158,7 @@ const createdClusterSchema = z.object({ const createdDropletSchema = z.strictObject({ id: z.number().int().positive(), name: nonEmptyString, - role: z.enum(["uut", "loadgen"]), + role: z.enum(["uut", "loadgen", "worker"]), publicIp: z.string().nullable(), privateIp: z.string().nullable() }); diff --git a/src/infra/types.ts b/src/infra/types.ts index d8981be..e0114e6 100644 --- a/src/infra/types.ts +++ b/src/infra/types.ts @@ -227,7 +227,7 @@ export interface SshIdentity { /** One image of perfkit.yaml, pinned: `ref` is `name:tag@digest`. */ export type ResolvedImage = Plan["images"][number]; -export type DropletRole = "uut" | "loadgen"; +export type DropletRole = "uut" | "loadgen" | "worker"; export type DropletPlan = Plan["uut"]; @@ -274,6 +274,8 @@ export interface ProvisionedRun extends RunResources { firewall: NonNullable; uut: CreatedDroplet & { publicIp: string; privateIp: string }; loadgen: CreatedDroplet & { publicIp: string; privateIp: string }; + /** The other load generators, in order; empty with `loadgen.count` 1. */ + workers: Array; /** The private key that logs in as root on both droplets. */ identity: SshIdentity; /** With a cluster in the plan. */ diff --git a/src/orchestrate/buildSippRoles.ts b/src/orchestrate/buildSippRoles.ts index 9cb5fe8..36a09b5 100644 --- a/src/orchestrate/buildSippRoles.ts +++ b/src/orchestrate/buildSippRoles.ts @@ -69,6 +69,8 @@ export interface SippRolesInput { cps?: number; /** A role's own address instead of `loadgenIp` (SIPp on the kind network). */ bindIps?: Partial>; + /** A uac instance's own address, when it runs on another load generator. */ + uacIps?: Record; } /** @@ -104,7 +106,9 @@ export function buildSippRoles(input: SippRolesInput): SippRoleSpec[] { "-sf", `${SIPP_PROJECT_DIR}/${scenario}`, "-i", - input.bindIps?.[role] ?? input.loadgenIp, + (role === "uac" ? input.uacIps?.[instance] : undefined) ?? + input.bindIps?.[role] ?? + input.loadgenIp, "-p", String(portOf(role) + instance - 1), "-cid_str", diff --git a/src/orchestrate/createAgentShell.ts b/src/orchestrate/createAgentShell.ts index 76d76c9..fa8d773 100644 --- a/src/orchestrate/createAgentShell.ts +++ b/src/orchestrate/createAgentShell.ts @@ -7,7 +7,7 @@ import { nameOf, RUN_LABEL } from "./constants.js"; import { demuxLogs, type EngineRequest } from "./createDockerEngine.js"; import type { RunState } from "./runState.schema.js"; import { shellQuote } from "./shellQuote.js"; -import type { HostShell } from "./types.js"; +import { type HostShell, workerIndexOf } from "./types.js"; const DOCKER_SOCKET = "/var/run/docker.sock"; const DEFAULT_TIMEOUT_MS = 10 * 60 * 1000; @@ -24,9 +24,10 @@ export function agentKeyPath(root: string): string { /** * The agent's way to run host commands. Each command runs in a short-lived * DOCKER_CLI_IMAGE container on the load generator (host network, the - * Docker socket and the run's root mounted at the same path). A UUT command - * goes over SSH to the UUT's private IP with the run-only key; with --local - * the UUT is this machine, so it runs the same way as a load-generator one. + * Docker socket and the run's root mounted at the same path). A UUT or worker + * (another load generator) command goes over SSH to its private IP with the + * run-only key; with --local the UUT is this machine, so it runs the same way + * as a load-generator one. */ export function createAgentShell(deps: { engine: EngineRequest; @@ -97,9 +98,23 @@ export function createAgentShell(deps: { return (host, command, options = {}) => { const timeoutMs = options.timeoutMs ?? DEFAULT_TIMEOUT_MS; - if (host === "uut" && state.mode === "digitalocean") { + const worker = workerIndexOf(host); + const remoteIp = + worker !== null + ? state.hosts.workers[worker - 1]?.privateIp + : host === "uut" + ? state.hosts.uutIp + : undefined; + if (worker !== null && remoteIp === undefined) { + return Promise.resolve({ + code: 255, + stdout: "", + stderr: `the run has no ${host}` + }); + } + if (remoteIp !== undefined && state.mode === "digitalocean") { return runHelper( - `ssh ${AGENT_SSH_OPTIONS} -i ${shellQuote(agentKeyPath(state.root))} root@${state.hosts.uutIp} ${shellQuote(command)}`, + `ssh ${AGENT_SSH_OPTIONS} -i ${shellQuote(agentKeyPath(state.root))} root@${remoteIp} ${shellQuote(command)}`, timeoutMs ); } diff --git a/src/orchestrate/createRunBenchmark.ts b/src/orchestrate/createRunBenchmark.ts index e600747..5199762 100644 --- a/src/orchestrate/createRunBenchmark.ts +++ b/src/orchestrate/createRunBenchmark.ts @@ -77,7 +77,14 @@ import type { RunState } from "./runState.schema.js"; import { buildOtelcolConfig, buildSamplerSpec } from "./sampler.js"; import { shellJoin, shellQuote, writeFileCommand } from "./shellQuote.js"; import type { LiveRecorder } from "../live/types.js"; -import type { AgentFs, Host, HostShell } from "./types.js"; +import { generatorOf as generatorFor, workerFileName } from "./workerFiles.js"; +import { AGENT_SSH_OPTIONS, agentKeyPath } from "./createAgentShell.js"; +import { + type AgentFs, + type Host, + type HostShell, + workerIndexOf +} from "./types.js"; export interface AgentDeps { shell: HostShell; @@ -126,6 +133,72 @@ export function createRunBenchmark(deps: AgentDeps, state: RunState) { options?: Parameters[2] ) => run("loadgen", kubeCommand(state, script, options), timeoutMs); + 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"; + const workerHosts = workers.map((_, index): Host => `worker-${index + 1}`); + /** `ssh` to a worker's private address, from the load generator. */ + const sshTo = (index: number) => + `ssh ${AGENT_SSH_OPTIONS} -i ${shellQuote(agentKeyPath(state.root))} root@${workers[index - 1]!.privateIp}`; + + /** Copies the named files of `rawDir` to the same directory on a worker. */ + const pushToWorker = async ( + index: number, + rawDir: string, + files: string[] + ) => { + const out = await run( + "loadgen", + `tar -C ${shellQuote(rawDir)} -cf - ${files.map((file) => shellQuote(file)).join(" ")} | ${sshTo(index)} ${shellQuote(`mkdir -p ${rawDir} && tar -C ${rawDir} -xf -`)}`, + 5 * 60 * 1000 + ); + if (out.code !== 0) { + throw new AgentFailure( + `could not copy the users to worker-${index}: ${lastLine(out.stderr, `exit ${out.code}`)}` + ); + } + }; + + /** + * Copies what the workers' SIPp processes wrote into `rawDir`, so the + * agent reads every load generator's files from one directory. SIPp names + * them after its pid, which two machines can share, so each worker's pids + * get the worker's number in front and are padded past any real pid: + * worker 1's pid 4242 is 100004242. + * + * @param strict - Fail when a worker can't be read; during a run it only + * means the next pull has more to copy + */ + const pullWorkers = async (rawDir: string, strict: boolean) => { + await Promise.all( + workers.map(async (_, position) => { + const index = position + 1; + const stage = `${rawDir}/.pull-${index}`; + await deps.fs.mkdir(stage); + const out = await run( + "loadgen", + `${sshTo(index)} ${shellQuote(`tar -C ${rawDir} -cf - .`)} | tar -C ${shellQuote(stage)} -xf -`, + 5 * 60 * 1000 + ); + if (out.code !== 0 && strict) { + throw new AgentFailure( + `could not read the results of worker-${index}: ${lastLine(out.stderr, `exit ${out.code}`)}` + ); + } + const names = await deps.fs.readdir(stage).catch(() => [] as string[]); + await Promise.all( + names.map(async (name) => { + const renamed = workerFileName(index, name); + if (renamed === null) return; + await deps.fs.rename(`${stage}/${name}`, `${rawDir}/${renamed}`); + }) + ); + }) + ); + }; + return async ( benchmark: BenchmarkConfig, rate?: BenchmarkRate @@ -156,14 +229,26 @@ export function createRunBenchmark(deps: AgentDeps, state: RunState) { name(`${benchmark.id}-sipp-${roleOf(role as SippRole, instance)}`) }; const instances = benchmark.load.instances; + const uacInstances = Array.from({ length: instances }, (_, 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) ), - ...Array.from({ length: instances }, (_, i) => - containers.sipp("uac", i + 1) - ) + ...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; 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` @@ -186,7 +271,8 @@ export function createRunBenchmark(deps: AgentDeps, state: RunState) { ...(rate ? { cps: rate.cps } : {}), ...(cluster && onCluster && cluster.fixedIps ? { bindIps: cluster.sippIps } - : {}) + : {}), + uacIps }); // Fails early, before anything starts, when the rate is missing. if (!onCluster) buildRoles(null); @@ -210,6 +296,11 @@ export function createRunBenchmark(deps: AgentDeps, state: RunState) { : run("uut", removeContainersCommand([containers.samplerUut])); const cleanup = async () => { + await Promise.all( + workerHosts.map((host) => + run(host, removeContainersCommand(onGenerator(host))) + ) + ); await run( "loadgen", removeContainersCommand([ @@ -254,12 +345,13 @@ export function createRunBenchmark(deps: AgentDeps, state: RunState) { .reverse() .map(({ role, instance }) => run( - "loadgen", + hostOf(role, instance), stopSippCommand(containers.sipp(role, instance)), 120_000 ) ) ); + await pullWorkers(rawDir, true); if (benchmark.telemetry.otlp) { await run( "loadgen", @@ -615,7 +707,7 @@ export function createRunBenchmark(deps: AgentDeps, state: RunState) { benchmark.load.transport === "tcp" ? SIPP_TCP_NOFILE : 1048576; const startRole = async (spec: SippRoleSpec) => { const out = await run( - "loadgen", + hostOf(spec.role, spec.instance), dockerRunCommand({ runId: state.runId, name: container(spec.role, spec.instance), @@ -668,6 +760,18 @@ export function createRunBenchmark(deps: AgentDeps, state: RunState) { ) ); } + // Each worker gets its own instances' shares of the uac users + 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 + ? pushToWorker(index, rawDir, own) + : run(`worker-${index}`, `mkdir -p ${shellQuote(rawDir)}`); + }) + ); await callees.reduce( (previous, spec) => previous.then(() => startRole(spec)), Promise.resolve() @@ -767,6 +871,7 @@ export function createRunBenchmark(deps: AgentDeps, state: RunState) { : await run("uut", "true", 20 * 1000); failedChecks = out.code === 0 ? 0 : failedChecks + 1; if (failedChecks >= MAX_FAILED_CHECKS) return false; + await pullWorkers(rawDir, false); } await deps.sleep(Math.max(1, Math.min(deadline, nextCheck) - deps.now())); return waitUntil(deadline); @@ -792,6 +897,7 @@ export function createRunBenchmark(deps: AgentDeps, state: RunState) { startedMs + (index + 1) * everyMs + STEP_GRACE_MS ); if (!ok) return { steps: Math.max(index, 1), aborted: unresponsive }; + await pullWorkers(rawDir, false); const evaluation = await evaluateCapacityStep({ fs: deps.fs, rawDir, diff --git a/src/orchestrate/createSshRemote.ts b/src/orchestrate/createSshRemote.ts index 964aed2..36a1a4c 100644 --- a/src/orchestrate/createSshRemote.ts +++ b/src/orchestrate/createSshRemote.ts @@ -15,9 +15,13 @@ const COPY_TIMEOUT_MS = 30 * 60 * 1000; export function createSshRemote(deps: { ssh: SshTools; privateKey: string; - publicIps: Record; + publicIps: Record; }): Remote { - const host = (role: Host) => deps.publicIps[role]; + const host = (role: Host) => { + const ip = deps.publicIps[role]; + if (!ip) throw new Error(`the run has no ${role} droplet`); + return ip; + }; const check = (what: string, out: { code: number; stderr: string }) => { if (out.code !== 0) { throw new Error(`${what}: ${lastLine(out.stderr, `exit ${out.code}`)}`); diff --git a/src/orchestrate/createStartRun.ts b/src/orchestrate/createStartRun.ts index 798b035..d110bf4 100644 --- a/src/orchestrate/createStartRun.ts +++ b/src/orchestrate/createStartRun.ts @@ -63,8 +63,11 @@ export function createStartRun(deps: StartRunDeps) { const fn = async (input: StartRunInput): Promise => { const paths = runPaths(input.root); + const workers: Host[] = input.hosts.workers.map( + (_, index) => `worker-${index + 1}` as const + ); const hosts: Host[] = - input.mode === "local" ? ["loadgen"] : ["uut", "loadgen"]; + input.mode === "local" ? ["loadgen"] : ["uut", "loadgen", ...workers]; const telemetry = input.plan.benchmarks.some((b) => b.telemetry.otlp); const staging = await mkdtemp(join(tmpdir(), "perfkit-stage-")); try { @@ -108,6 +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 + 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}` + ) + ) + ); const info = async (host: Host): Promise => parseDockerInfo( @@ -170,6 +190,7 @@ 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/runState.schema.ts b/src/orchestrate/runState.schema.ts index 634e883..8b33979 100644 --- a/src/orchestrate/runState.schema.ts +++ b/src/orchestrate/runState.schema.ts @@ -79,6 +79,10 @@ export const runStateSchema = z.strictObject({ hosts: z.strictObject({ uutIp: z.ipv4(), loadgenIp: z.ipv4(), + /** The other load generators (`infra.loadgen.count` - 1), in order. */ + workers: z + .array(z.strictObject({ publicIp: z.ipv4(), privateIp: z.ipv4() })) + .default([]), vpcCidr: nonEmptyString }), machines: z.strictObject({ diff --git a/src/orchestrate/types.ts b/src/orchestrate/types.ts index 54f278e..08e8352 100644 --- a/src/orchestrate/types.ts +++ b/src/orchestrate/types.ts @@ -3,9 +3,16 @@ */ import type { SshResult } from "../infra/index.js"; -export type Host = "uut" | "loadgen"; +/** `worker-` is the n-th load generator after the first (`loadgen`). */ +export type Host = "uut" | "loadgen" | `worker-${number}`; -/** Runs a shell command as root on one of the run's two hosts. */ +/** The index (from 1) of a worker host, or null for the other hosts. */ +export function workerIndexOf(host: Host): number | null { + const match = /^worker-(\d+)$/.exec(host); + return match ? Number(match[1]) : null; +} + +/** Runs a shell command as root on one of the run's hosts. */ export type HostShell = ( host: Host, command: string, diff --git a/src/orchestrate/workerFiles.ts b/src/orchestrate/workerFiles.ts new file mode 100644 index 0000000..759f3a9 --- /dev/null +++ b/src/orchestrate/workerFiles.ts @@ -0,0 +1,27 @@ +/** + * Copyright (C) 2026 by Fonoster Inc (https://fonoster.com). MIT License. + */ +import type { Host } from "./types.js"; + +/** + * The load generator a uac instance runs on: round robin over the first + * (`loadgen`) and its `workers`, so instance 1 is always on `loadgen`. + */ +export function generatorOf(instance: number, workers: number): Host { + const slot = (instance - 1) % (workers + 1); + return slot === 0 ? "loadgen" : `worker-${slot}`; +} + +/** + * 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). + */ +export function workerFileName(worker: number, name: string): string | null { + const match = /^(.+)_(\d+)_(.*)$/.exec(name); + if (!match) return null; + const [, base, pid, rest] = match; + return `${base}_${worker}${(pid as string).padStart(8, "0")}_${rest}`; +} diff --git a/src/schemas/perfkitConfig.schema.ts b/src/schemas/perfkitConfig.schema.ts index 64ec19a..7535793 100644 --- a/src/schemas/perfkitConfig.schema.ts +++ b/src/schemas/perfkitConfig.schema.ts @@ -94,11 +94,25 @@ const configClusterSchema = z.strictObject({ version: nonEmptyString.default("latest") }); +/** + * The load generator: `count` droplets of `size`. The first runs the agent and + * the callees; the others only add uac SIPp processes, each with its own address + * and its own pool of source ports (over TCP every call opens a connection). + */ +const loadgenSchema = machineSizeSchema.extend({ + count: z + .number() + .int({ error: "expected a whole number" }) + .min(1, { error: "expected 1 load generator or more" }) + .max(8, { error: "expected 8 load generators or fewer" }) + .default(1) +}); + const configInfraSchema = z.strictObject({ provider: z.literal("digitalocean").default("digitalocean"), region: nonEmptyString, uut: machineSizeSchema, - loadgen: machineSizeSchema, + loadgen: loadgenSchema, /** Required when a benchmark has `uut.helm`. */ cluster: configClusterSchema.optional(), maxLifetime: durationSchema.default("9h") diff --git a/test/infra/createPlanRun.unit.test.ts b/test/infra/createPlanRun.unit.test.ts index a361f53..44ec8cc 100644 --- a/test/infra/createPlanRun.unit.test.ts +++ b/test/infra/createPlanRun.unit.test.ts @@ -71,6 +71,33 @@ describe("createPlanRun", () => { ]); }); + it("should plan no worker with one load generator and count - 1 with more", async () => { + // Arrange + const { planRun } = setup(); + const config = loadConfig(); + const planFor = (count: number) => { + config.infra.loadgen.count = count; + return planRun({ + ...input(), + config: perfkitConfigSchema.parse(config) + }); + }; + + // Act + const [one, three] = [await planFor(1), await planFor(3)]; + + // Assert + expect(one.workers).to.deep.equal([]); + expect(three.workers).to.deep.equal( + [1, 2].map((n) => ({ + role: "worker", + name: `perfkit-${RUN_ID}-worker-${n}`, + size: "c-16", + image: "ubuntu-22-04-x64" + })) + ); + }); + it("should keep only the benchmarks asked for", async () => { // Arrange const { planRun } = setup(); diff --git a/test/infra/createProvision.unit.test.ts b/test/infra/createProvision.unit.test.ts index 50804d7..856216c 100644 --- a/test/infra/createProvision.unit.test.ts +++ b/test/infra/createProvision.unit.test.ts @@ -115,6 +115,43 @@ describe("createProvision", () => { ); }); + it("should create the other load generators, wait for them and authorize the agent's key on them", async () => { + // Arrange + const { provision, client, ssh } = setup(); + const plan = await makePlan(2); + + // Act + const run = await provision(plan, { authorizedKeys: [AGENT_KEY] }); + + // Assert + const created = client.callsTo("createDroplet").map(([input]) => ({ + name: (input as { name: string }).name, + userData: (input as { user_data: string }).user_data + })); + expect(created.map(({ name }) => name)).to.deep.equal([ + `perfkit-${RUN_ID}-uut`, + `perfkit-${RUN_ID}-loadgen`, + `perfkit-${RUN_ID}-worker-1` + ]); + expect(created[2]!.userData).to.include(AGENT_KEY); + expect( + run.workers.map(({ name, role }) => ({ name, role })) + ).to.deep.equal([{ name: `perfkit-${RUN_ID}-worker-1`, role: "worker" }]); + expect(run.workers[0]!.privateIp).to.match(/^10\.116\.0\./); + expect(ssh.commands).to.have.length(3); + }); + + it("should return no workers with one load generator", async () => { + // Arrange + const { provision } = setup(); + + // Act + const run = await provision(await makePlan()); + + // Assert + expect(run.workers).to.deep.equal([]); + }); + it("should retry SSH until it is up", async () => { // Arrange const { now, sleep } = fakeClock(); diff --git a/test/infra/createSweep.unit.test.ts b/test/infra/createSweep.unit.test.ts index 5913c98..05a2069 100644 --- a/test/infra/createSweep.unit.test.ts +++ b/test/infra/createSweep.unit.test.ts @@ -32,6 +32,37 @@ describe("createSweep", () => { expect(teardown.callCount).to.equal(0); }); + it("should sweep a run's worker load generators like its other droplets", async () => { + // Arrange + const client = new FakeDigitalOcean(); + const runId = "20261103T021210Z-41d9a2c"; + ["uut", "loadgen", "worker-1", "worker-2"].forEach((role) => + client.seedDroplet({ + name: `perfkit-${runId}-${role}`, + tags: ["perfkit", `perfkit-run-${runId}`], + createdAt: START - 10 * 3600 * 1000 + }) + ); + const teardown = sinon.stub().resolves({ deleted: [], failures: [] }); + const sweep = createSweep({ client, teardown, now: () => START }); + + // Act + const result = await sweep({ olderThan: "9h" }); + + // Assert + expect(result.runs).to.have.length(1); + const resources = teardown.firstCall.args[0] as { + droplets: Array<{ name: string; role: string }>; + }; + expect(resources.droplets.map(({ name }) => name)).to.have.length(4); + expect(resources.droplets.map(({ role }) => role)).to.have.members([ + "uut", + "loadgen", + "worker", + "worker" + ]); + }); + it("should look only at droplets tagged perfkit", async () => { // Arrange const client = new FakeDigitalOcean(); diff --git a/test/infra/fakes.ts b/test/infra/fakes.ts index ec41bd7..7c8370c 100644 --- a/test/infra/fakes.ts +++ b/test/infra/fakes.ts @@ -440,7 +440,7 @@ export async function perfkit( } /** The plan of the Routr example at 2.31.5, built by the real planRun. */ -export async function makePlan(): Promise { +export async function makePlan(loadgenCount = 1): Promise { const { parse } = await import("yaml"); const { readFileSync } = await import("node:fs"); const { ROUTR_CONFIG } = await import("../fixtures.js"); @@ -453,10 +453,10 @@ export async function makePlan(): Promise { }), getPublicIp: async () => "203.0.113.7" }); + const raw = parse(readFileSync(ROUTR_CONFIG, "utf8")); + raw.infra.loadgen.count = loadgenCount; return planRun({ - config: perfkitConfigSchema.parse( - parse(readFileSync(ROUTR_CONFIG, "utf8")) - ), + config: perfkitConfigSchema.parse(raw), targetVersion: "2.31.5", commit: COMMIT, startedAt: "2026-11-03T14:12:10Z" diff --git a/test/orchestrate/buildSippRoles.unit.test.ts b/test/orchestrate/buildSippRoles.unit.test.ts index ca9bbd0..c9f39fe 100644 --- a/test/orchestrate/buildSippRoles.unit.test.ts +++ b/test/orchestrate/buildSippRoles.unit.test.ts @@ -492,6 +492,32 @@ describe("buildSippRoles with several uac instances", () => { registerUsers: 200 }); + it("should bind an instance on another load generator to that machine's address", () => { + // Arrange + const benchmark = capacityWithCallees({ instances: 3 }); + + // Act + const roles = buildSippRoles({ + ...input(benchmark), + uacIps: { 2: "10.0.0.9" } + }); + + // Assert + const bound = (instance: number) => { + const spec = roles.find( + (role) => role.role === "uac" && role.instance === instance + ); + return spec!.args[spec!.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" + ]); + const callee = roles.find(({ role }) => role === "register"); + expect(callee!.args[callee!.args.indexOf("-i") + 1]).to.equal("10.0.0.3"); + }); + it("should split a capacity ramp's rate and open calls among the instances", () => { // Arrange const benchmark = capacityWithCallees({ diff --git a/test/orchestrate/workerFiles.unit.test.ts b/test/orchestrate/workerFiles.unit.test.ts new file mode 100644 index 0000000..8c1a6cd --- /dev/null +++ b/test/orchestrate/workerFiles.unit.test.ts @@ -0,0 +1,63 @@ +/** + * Copyright (C) 2026 by Fonoster Inc (https://fonoster.com). MIT License. + */ +import { expect } from "chai"; +import { + generatorOf, + workerFileName +} from "../../src/orchestrate/workerFiles.js"; + +describe("generatorOf", () => { + it("should run everything on the load generator without workers", () => { + expect([1, 2, 3].map((n) => generatorOf(n, 0))).to.deep.equal([ + "loadgen", + "loadgen", + "loadgen" + ]); + }); + + it("should spread instances round robin, the first on the load generator", () => { + expect([1, 2, 3, 4, 5].map((n) => generatorOf(n, 1))).to.deep.equal([ + "loadgen", + "worker-1", + "loadgen", + "worker-1", + "loadgen" + ]); + expect([1, 2, 3, 4].map((n) => generatorOf(n, 2))).to.deep.equal([ + "loadgen", + "worker-1", + "worker-2", + "loadgen" + ]); + }); +}); + +describe("workerFileName", () => { + it("should put the worker's number in front of the padded pid", () => { + expect(workerFileName(1, "uac_invite_4242_.csv")).to.equal( + "uac_invite_100004242_.csv" + ); + expect(workerFileName(2, "uac_invite_4242_rtt.csv")).to.equal( + "uac_invite_200004242_rtt.csv" + ); + expect(workerFileName(1, "uac_invite_4242_error_codes.csv")).to.equal( + "uac_invite_100004242_error_codes.csv" + ); + }); + + 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]!; + expect(pid(workerFileName(1, "uac_invite_1_.csv")!).length).to.be.at.least( + 9 + ); + expect(workerFileName(1, "uac_invite_7_.csv")).to.not.equal( + workerFileName(2, "uac_invite_7_.csv") + ); + }); + + it("should leave out files that are not SIPp's", () => { + expect(workerFileName(1, "users-uac-2.csv")).to.equal(null); + }); +}); diff --git a/test/schemas/perfkitConfig.unit.test.ts b/test/schemas/perfkitConfig.unit.test.ts index 96d1659..f794cbd 100644 --- a/test/schemas/perfkitConfig.unit.test.ts +++ b/test/schemas/perfkitConfig.unit.test.ts @@ -124,6 +124,20 @@ describe("perfkitConfigSchema", () => { }); }); + it("should default infra.loadgen.count to 1 and accept 1 to 8", () => { + // Arrange + const raw = loadConfig(); + + // Act and Assert + expect(perfkitConfigSchema.parse(raw).infra.loadgen.count).to.equal(1); + raw.infra.loadgen.count = 2; + expect(perfkitConfigSchema.parse(raw).infra.loadgen.count).to.equal(2); + [0, 9, 1.5].forEach((bad) => { + raw.infra.loadgen.count = bad; + expect(() => perfkitConfigSchema.parse(raw)).to.throw(); + }); + }); + it("should default a capacity warm-up to 0s and accept one shorter than a step", () => { const raw = loadConfig(); expect(perfkitConfigSchema.parse(raw).benchmarks[0]?.warmup).to.equal(