From d00306f9256ab61cc9e3d1572e8338273e46809d Mon Sep 17 00:00:00 2001 From: kjgbot Date: Tue, 21 Jul 2026 01:08:57 +0200 Subject: [PATCH 1/3] add in-cluster load harness with SLO evidence --- .github/workflows/ci.yml | 39 +++ docs/load-harness.md | 68 ++++ package-lock.json | 35 ++ package.json | 2 + src/environments/k6-job.ts | 440 ++++++++++++++++++++++++++ src/environments/load-harness.test.ts | 231 ++++++++++++++ src/environments/load-harness.ts | 272 ++++++++++++++++ src/environments/load-profile.ts | 143 +++++++++ src/index.ts | 52 +++ test/e2e/load-harness.e2e.ts | 151 +++++++++ 10 files changed, 1433 insertions(+) create mode 100644 docs/load-harness.md create mode 100644 src/environments/k6-job.ts create mode 100644 src/environments/load-harness.test.ts create mode 100644 src/environments/load-harness.ts create mode 100644 src/environments/load-profile.ts create mode 100644 test/e2e/load-harness.e2e.ts diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 067f67d..883e20a 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -52,3 +52,42 @@ jobs: - name: Verify package contents run: npm pack --dry-run + + load-e2e: + runs-on: ubuntu-latest + timeout-minutes: 15 + permissions: + contents: read + steps: + - name: Checkout + uses: actions/checkout@v6 + + - name: Setup Node + uses: actions/setup-node@v6 + with: + node-version: 20 + cache: npm + + - name: Install dependencies + run: npm ci + + - name: Install kind + run: | + curl --fail --location --output kind https://kind.sigs.k8s.io/dl/v0.31.0/kind-linux-amd64 + echo 'eb244cbafcc157dff60cf68693c14c9a75c4e6e6fedaf9cd71c58117cb93e3fa kind' | sha256sum --check + chmod +x kind + sudo mv kind /usr/local/bin/kind + + - name: Run in-cluster load harness E2E + run: | + trap 'kind delete cluster --name factory-load-e2e' EXIT + kind create cluster --name factory-load-e2e --image kindest/node:v1.35.0@sha256:452d707d4862f52530247495d180205e029056831160e22870e37e3f6c1ac31f + npm run test:e2e:load + + - name: Upload load evidence + if: always() + uses: actions/upload-artifact@v6 + with: + name: factory-load-evidence-${{ github.event.pull_request.head.sha || github.sha }} + path: artifacts/load-e2e/*.json + if-no-files-found: error diff --git a/docs/load-harness.md b/docs/load-harness.md new file mode 100644 index 0000000..093b628 --- /dev/null +++ b/docs/load-harness.md @@ -0,0 +1,68 @@ +# In-cluster load harness + +`runLoad(environment, profile)` creates a k6 Job in the environment's Kubernetes +namespace, waits for it to complete, captures its summary, and returns a +structured SLO verdict. The environment endpoint map should contain URLs that +are reachable from a pod; Kubernetes Service DNS names are preferred when the +target stack runs in the same cluster. + +```ts +import { runLoad } from '@agent-relay/factory' + +const result = await runLoad( + { + id: 'verify-123', + namespace: 'verify-123', + endpoints: { api: 'http://api.verify-123.svc.cluster.local:8080' }, + }, + { + name: 'api-load', + targets: [ + { name: 'read', endpoint: 'api', path: '/items', weight: 4 }, + { + name: 'write', + endpoint: 'api', + path: '/items', + method: 'POST', + body: { value: 'example' }, + }, + ], + vus: 50, + duration: '30s', + ramp: { up: '5s', down: '2s' }, + thresholds: { + maxP95LatencyMs: 2_000, + maxP99LatencyMs: 5_000, + maxErrorRate: 0.01, + minThroughputRps: 10, + }, + }, + { evidencePath: 'artifacts/load.json' }, +) + +if (!result.passed) { + console.error(result.violations) +} +``` + +Set `rps` to use an arrival-rate scenario; `vus` then controls preallocated +workers and `maxVus` caps scaling. Without `rps`, k6 runs a VU-based scenario. +Profiles can also be checked in as JSON or YAML and loaded with +`loadLoadProfile(path)`. + +The returned `evidence` and `evidenceJson` include request count, p95 and p99, +error rate, throughput, a cumulative latency histogram, thresholds, and every +violated metric. Request headers and bodies are deliberately omitted from +evidence. The ephemeral ConfigMap and Job are deleted by default; pass +`cleanup: false` when a surrounding environment teardown owns cleanup. + +Run the real kind-cluster proof with: + +```bash +kind create cluster --name factory-load-e2e +npm run test:e2e:load +kind delete cluster --name factory-load-e2e +``` + +The E2E deploys a healthy service, drives a 50-VU load by default, records its +evidence, asserts a lenient SLO passes, and asserts a strict p95 SLO fails. diff --git a/package-lock.json b/package-lock.json index 65285cd..94a89a4 100644 --- a/package-lock.json +++ b/package-lock.json @@ -30,6 +30,7 @@ "@types/proper-lockfile": "^4.1.4", "esbuild": "^0.28.1", "tsc-alias": "^1.8.17", + "tsx": "4.23.1", "typescript": "^5.7.0", "vitest": "^4.1.8" }, @@ -6897,6 +6898,40 @@ "integrity": "sha512-oJFu94HQb+KVduSUQL7wnpmqnfmLsOA/nAh6b6EH0wCEoK0/mPeXU6c3wKDV83MkOuHPRHtSXKKU99IBazS/2w==", "license": "0BSD" }, + "node_modules/tsx": { + "version": "4.23.1", + "resolved": "https://registry.npmjs.org/tsx/-/tsx-4.23.1.tgz", + "integrity": "sha512-GQHnkIfxyx1wYCOS/wonik5MVRZU9hi1TEZmzGZSCJB1y9YgoZ8H6itNE/u4suE+yLmOzuE4E5S4TZ/ZX2wcWQ==", + "dev": true, + "license": "MIT", + "dependencies": { + "esbuild": "~0.28.0" + }, + "bin": { + "tsx": "dist/cli.mjs" + }, + "engines": { + "node": ">=18.0.0" + }, + "optionalDependencies": { + "fsevents": "~2.3.3" + } + }, + "node_modules/tsx/node_modules/fsevents": { + "version": "2.3.3", + "resolved": "https://registry.npmjs.org/fsevents/-/fsevents-2.3.3.tgz", + "integrity": "sha512-5xoDfX+fL7faATnagmWPpbFtwh/R77WmMMqqHGS65C3vvB0YHrgF+B1YmZ3441tMj5n63k0212XNoJwzlhffQw==", + "dev": true, + "hasInstallScript": true, + "license": "MIT", + "optional": true, + "os": [ + "darwin" + ], + "engines": { + "node": "^8.16.0 || ^10.6.0 || >=11.0.0" + } + }, "node_modules/tweetnacl": { "version": "0.14.5", "resolved": "https://registry.npmjs.org/tweetnacl/-/tweetnacl-0.14.5.tgz", diff --git a/package.json b/package.json index 41cc3fd..e4309fd 100644 --- a/package.json +++ b/package.json @@ -65,6 +65,7 @@ "build": "tsc -p tsconfig.build.json && tsc-alias -p tsconfig.build.json", "featuremap:check": "node bin/factory.mjs featuremap check", "test": "vitest run", + "test:e2e:load": "tsx test/e2e/load-harness.e2e.ts", "verify:e2e": "node scripts/verify-packed-e2e.mjs", "factory-build": "factory-build/run-factory-build.sh" }, @@ -91,6 +92,7 @@ "@types/proper-lockfile": "^4.1.4", "esbuild": "^0.28.1", "tsc-alias": "^1.8.17", + "tsx": "4.23.1", "typescript": "^5.7.0", "vitest": "^4.1.8" } diff --git a/src/environments/k6-job.ts b/src/environments/k6-job.ts new file mode 100644 index 0000000..e45823d --- /dev/null +++ b/src/environments/k6-job.ts @@ -0,0 +1,440 @@ +import { spawn } from 'node:child_process' +import { setTimeout as delay } from 'node:timers/promises' + +import type { ResolvedLoadProfile, ResolvedLoadTargetProfile } from './load-profile.js' +import { durationToMilliseconds } from './load-profile.js' + +export const DEFAULT_K6_IMAGE = 'grafana/k6:1.7.1@sha256:4fd3a694926b064d3491d9b02b01cde886583c4931f1223816e3d9a7bdfa7e0f' +export const K6_EVIDENCE_PREFIX = 'FACTORY_LOAD_EVIDENCE_JSON=' + +export interface LoadEnvironment { + id: string + endpoints: Record + namespace?: string + kubeContext?: string +} + +export interface ResolvedLoadTarget { + name: string + url: string + method: ResolvedLoadTargetProfile['method'] + headers: Record + body: string | null + expectedStatuses: number[] | null + weight: number +} + +export interface K6LoadJobResources { + jobName: string + configMapName: string + namespace: string + timeoutMs: number + resources: Array> +} + +export interface KubernetesLoadJobClient { + apply(resources: Array>, namespace: string): Promise + waitForCompletion(jobName: string, namespace: string, timeoutMs: number): Promise + logs(jobName: string, namespace: string): Promise + delete(jobName: string, configMapName: string, namespace: string): Promise +} + +export interface KubectlCommandResult { + stdout: string + stderr: string +} + +export type KubectlCommandRunner = ( + args: string[], + input?: string, +) => Promise + +export interface KubectlLoadJobClientOptions { + context?: string + executable?: string + pollIntervalMs?: number + runner?: KubectlCommandRunner +} + +const commandError = ( + executable: string, + args: string[], + code: number | null, + stderr: string, +): Error => new Error( + `${executable} ${args.join(' ')} exited with ${code ?? 'no status'}: ${stderr.trim().slice(-4_000) || 'no stderr'}`, +) + +export const defaultKubectlCommandRunner = ( + executable = 'kubectl', +): KubectlCommandRunner => async (args, input) => await new Promise((resolve, reject) => { + const child = spawn(executable, args, { stdio: ['pipe', 'pipe', 'pipe'] }) + let stdout = '' + let stderr = '' + + child.stdout.setEncoding('utf8') + child.stderr.setEncoding('utf8') + child.stdout.on('data', (chunk: string) => { stdout += chunk }) + child.stderr.on('data', (chunk: string) => { stderr += chunk }) + child.once('error', reject) + child.once('close', (code) => { + if (code === 0) resolve({ stdout, stderr }) + else reject(commandError(executable, args, code, stderr)) + }) + child.stdin.end(input) +}) + +export class KubectlLoadJobClient implements KubernetesLoadJobClient { + private readonly contextArgs: string[] + private readonly pollIntervalMs: number + private readonly runner: KubectlCommandRunner + + constructor(options: KubectlLoadJobClientOptions = {}) { + this.contextArgs = options.context ? ['--context', options.context] : [] + this.pollIntervalMs = options.pollIntervalMs ?? 1_000 + this.runner = options.runner ?? defaultKubectlCommandRunner(options.executable) + } + + async apply(resources: Array>, namespace: string): Promise { + await this.runner([ + ...this.contextArgs, + '--namespace', namespace, + 'apply', '--filename', '-', + ], JSON.stringify({ apiVersion: 'v1', kind: 'List', items: resources })) + } + + async waitForCompletion(jobName: string, namespace: string, timeoutMs: number): Promise { + const deadline = Date.now() + timeoutMs + while (Date.now() < deadline) { + const { stdout } = await this.runner([ + ...this.contextArgs, + '--namespace', namespace, + 'get', 'job', jobName, + '--output', 'json', + ]) + const job = JSON.parse(stdout) as { + status?: { + succeeded?: number + failed?: number + conditions?: Array<{ type?: string; status?: string; reason?: string; message?: string }> + } + } + if ((job.status?.succeeded ?? 0) > 0) return + if ((job.status?.failed ?? 0) > 0 || job.status?.conditions?.some( + (condition) => condition.type === 'Failed' && condition.status === 'True', + )) { + const failure = job.status?.conditions?.find( + (condition) => condition.type === 'Failed' && condition.status === 'True', + ) + throw new Error( + `k6 Job ${namespace}/${jobName} failed${failure?.reason ? ` (${failure.reason})` : ''}${failure?.message ? `: ${failure.message}` : ''}`, + ) + } + await delay(Math.min(this.pollIntervalMs, Math.max(1, deadline - Date.now()))) + } + throw new Error(`Timed out after ${timeoutMs}ms waiting for k6 Job ${namespace}/${jobName}`) + } + + async logs(jobName: string, namespace: string): Promise { + const { stdout } = await this.runner([ + ...this.contextArgs, + '--namespace', namespace, + 'logs', `job/${jobName}`, + '--container', 'k6', + ]) + return stdout + } + + async delete(jobName: string, configMapName: string, namespace: string): Promise { + await this.runner([ + ...this.contextArgs, + '--namespace', namespace, + 'delete', + `job/${jobName}`, + `configmap/${configMapName}`, + '--ignore-not-found=true', + '--wait=false', + ]) + } +} + +export function resolveLoadTargets( + environment: LoadEnvironment, + profile: ResolvedLoadProfile, +): ResolvedLoadTarget[] { + return profile.targets.map((target) => { + const baseUrl = target.url ?? environment.endpoints[target.endpoint as string] + if (!baseUrl) { + throw new Error( + `Load target ${target.name} references unknown environment endpoint ${target.endpoint}`, + ) + } + + let url: URL + try { + const normalizedBase = baseUrl.endsWith('/') ? baseUrl : `${baseUrl}/` + url = new URL(target.path ?? '', normalizedBase) + } catch { + throw new Error(`Load target ${target.name} resolved to an invalid URL`) + } + if (url.protocol !== 'http:' && url.protocol !== 'https:') { + throw new Error(`Load target ${target.name} must resolve to http or https`) + } + + const headers = { ...(target.headers ?? {}) } + let body: string | null = null + if (target.body !== undefined) { + if (typeof target.body === 'string') { + body = target.body + } else { + try { + body = JSON.stringify(target.body) + } catch { + throw new Error(`Load target ${target.name} body is not JSON serializable`) + } + const hasContentType = Object.keys(headers).some((header) => header.toLowerCase() === 'content-type') + if (!hasContentType) headers['content-type'] = 'application/json' + } + } + + return { + name: target.name, + url: url.toString(), + method: target.method, + headers, + body, + expectedStatuses: target.expectedStatuses ?? null, + weight: target.weight, + } + }) +} + +export function k6ScenarioFor(profile: ResolvedLoadProfile): Record { + const target = profile.rps ?? profile.vus as number + if (profile.rps !== undefined) { + const preAllocatedVUs = profile.vus ?? Math.max(1, Math.ceil(profile.rps)) + const maxVUs = profile.maxVus ?? Math.max(preAllocatedVUs, Math.ceil(preAllocatedVUs * 2)) + if (profile.ramp) { + return { + executor: 'ramping-arrival-rate', + startRate: 0, + timeUnit: '1s', + preAllocatedVUs, + maxVUs, + stages: [ + { duration: profile.ramp.up, target }, + { duration: profile.duration, target }, + ...(profile.ramp.down ? [{ duration: profile.ramp.down, target: 0 }] : []), + ], + gracefulStop: '5s', + } + } + return { + executor: 'constant-arrival-rate', + rate: profile.rps, + timeUnit: '1s', + duration: profile.duration, + preAllocatedVUs, + maxVUs, + gracefulStop: '5s', + } + } + + if (profile.ramp) { + return { + executor: 'ramping-vus', + startVUs: 0, + stages: [ + { duration: profile.ramp.up, target }, + { duration: profile.duration, target }, + ...(profile.ramp.down ? [{ duration: profile.ramp.down, target: 0 }] : []), + ], + gracefulRampDown: '5s', + } + } + return { + executor: 'constant-vus', + vus: profile.vus, + duration: profile.duration, + gracefulStop: '5s', + } +} + +export function renderK6Script( + environment: LoadEnvironment, + profile: ResolvedLoadProfile, +): string { + const targets = resolveLoadTargets(environment, profile) + const scenario = k6ScenarioFor(profile) + return `import http from 'k6/http'; +import { Counter } from 'k6/metrics'; + +const targets = ${JSON.stringify(targets)}; +const histogramBucketsMs = ${JSON.stringify(profile.histogramBucketsMs)}; +const requests = new Counter('factory_requests'); +const errors = new Counter('factory_errors'); +const latencyMetricNames = histogramBucketsMs.map((_, index) => 'factory_latency_bucket_' + index); +const latencyCounters = latencyMetricNames.map((name) => new Counter(name)); +const latencyOverflow = new Counter('factory_latency_overflow'); +const totalWeight = targets.reduce((sum, target) => sum + target.weight, 0); + +export const options = ${JSON.stringify({ + scenarios: { factory_load: scenario }, + summaryTrendStats: ['avg', 'min', 'med', 'max', 'p(90)', 'p(95)', 'p(99)'], + })}; + +function selectTarget() { + let selection = Math.random() * totalWeight; + for (const target of targets) { + selection -= target.weight; + if (selection < 0) return target; + } + return targets[targets.length - 1]; +} + +function metric(data, name, key) { + const value = data.metrics[name] && data.metrics[name].values[key]; + return typeof value === 'number' && Number.isFinite(value) ? value : 0; +} + +export default function () { + const target = selectTarget(); + const response = http.request(target.method, target.url, target.body, { + headers: target.headers, + tags: { factory_target: target.name }, + }); + const latency = response.timings.duration; + const failed = target.expectedStatuses === null + ? response.status < 200 || response.status >= 400 + : !target.expectedStatuses.includes(response.status); + requests.add(1); + errors.add(failed ? 1 : 0); + for (let index = 0; index < histogramBucketsMs.length; index += 1) { + latencyCounters[index].add(latency <= histogramBucketsMs[index] ? 1 : 0); + } + latencyOverflow.add(latency > histogramBucketsMs[histogramBucketsMs.length - 1] ? 1 : 0); +} + +export function handleSummary(data) { + const requestCount = metric(data, 'factory_requests', 'count'); + const errorCount = metric(data, 'factory_errors', 'count'); + const summary = { + requestCount, + errorCount, + errorRate: requestCount > 0 ? errorCount / requestCount : 1, + throughputRps: metric(data, 'factory_requests', 'rate'), + durationMs: data.state && typeof data.state.testRunDurationMs === 'number' + ? data.state.testRunDurationMs + : 0, + latency: { + minMs: metric(data, 'http_req_duration', 'min'), + averageMs: metric(data, 'http_req_duration', 'avg'), + medianMs: metric(data, 'http_req_duration', 'med'), + maxMs: metric(data, 'http_req_duration', 'max'), + p95Ms: metric(data, 'http_req_duration', 'p(95)'), + p99Ms: metric(data, 'http_req_duration', 'p(99)'), + }, + histogram: histogramBucketsMs.map((bucket, index) => ({ + upperBoundMs: bucket, + count: metric(data, latencyMetricNames[index], 'count'), + })).concat([{ + upperBoundMs: null, + count: metric(data, 'factory_latency_overflow', 'count'), + }]), + }; + return { stdout: '${K6_EVIDENCE_PREFIX}' + JSON.stringify(summary) + '\\n' }; +} +` +} + +const dnsLabel = (value: string): string => { + const normalized = value.toLowerCase() + .replace(/[^a-z0-9-]+/gu, '-') + .replace(/^-+|-+$/gu, '') + .slice(0, 40) + return normalized || 'load' +} + +export interface CreateK6LoadJobOptions { + jobName: string + namespace: string + image?: string + timeoutMs?: number +} + +export function createK6LoadJobResources( + environment: LoadEnvironment, + profile: ResolvedLoadProfile, + options: CreateK6LoadJobOptions, +): K6LoadJobResources { + const jobName = dnsLabel(options.jobName).slice(0, 63) + const configMapName = `${jobName}-script`.slice(0, 63) + const workloadDurationMs = durationToMilliseconds(profile.duration) + + (profile.ramp ? durationToMilliseconds(profile.ramp.up) : 0) + + (profile.ramp?.down ? durationToMilliseconds(profile.ramp.down) : 0) + const timeoutMs = options.timeoutMs ?? workloadDurationMs + 120_000 + const labels = { + 'app.kubernetes.io/name': 'factory-load', + 'app.kubernetes.io/component': 'load-generator', + 'factory.agent-relay.dev/environment': dnsLabel(environment.id), + 'factory.agent-relay.dev/profile': dnsLabel(profile.name), + } + + return { + jobName, + configMapName, + namespace: options.namespace, + timeoutMs, + resources: [ + { + apiVersion: 'v1', + kind: 'ConfigMap', + metadata: { name: configMapName, namespace: options.namespace, labels }, + immutable: true, + data: { 'load.js': renderK6Script(environment, profile) }, + }, + { + apiVersion: 'batch/v1', + kind: 'Job', + metadata: { name: jobName, namespace: options.namespace, labels }, + spec: { + backoffLimit: 0, + activeDeadlineSeconds: Math.max(1, Math.ceil(timeoutMs / 1_000)), + ttlSecondsAfterFinished: 600, + template: { + metadata: { labels }, + spec: { + restartPolicy: 'Never', + automountServiceAccountToken: false, + enableServiceLinks: false, + securityContext: { + runAsNonRoot: true, + seccompProfile: { type: 'RuntimeDefault' }, + }, + containers: [{ + name: 'k6', + image: options.image ?? DEFAULT_K6_IMAGE, + imagePullPolicy: 'IfNotPresent', + args: ['run', '--quiet', '/scripts/load.js'], + env: [{ name: 'K6_NO_USAGE_REPORT', value: 'true' }], + securityContext: { + allowPrivilegeEscalation: false, + capabilities: { drop: ['ALL'] }, + }, + resources: { + requests: { cpu: '100m', memory: '64Mi' }, + limits: { cpu: '1', memory: '512Mi' }, + }, + volumeMounts: [{ name: 'script', mountPath: '/scripts', readOnly: true }], + }], + volumes: [{ + name: 'script', + configMap: { name: configMapName }, + }], + }, + }, + }, + }, + ], + } +} diff --git a/src/environments/load-harness.test.ts b/src/environments/load-harness.test.ts new file mode 100644 index 0000000..4720aa0 --- /dev/null +++ b/src/environments/load-harness.test.ts @@ -0,0 +1,231 @@ +import { mkdtemp, rm, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' + +import { afterEach, describe, expect, it } from 'vitest' + +import { + evaluateLoadSlo, + parseK6LoadMeasurements, + runLoad, + type LoadMeasurements, +} from './load-harness' +import { + createK6LoadJobResources, + K6_EVIDENCE_PREFIX, + k6ScenarioFor, + type KubernetesLoadJobClient, +} from './k6-job' +import { loadLoadProfile, LoadProfileSchema, type LoadProfile } from './load-profile' + +const measurements: LoadMeasurements = { + requestCount: 1_000, + errorCount: 2, + errorRate: 0.002, + throughputRps: 99.5, + durationMs: 10_050, + latency: { + minMs: 2, + averageMs: 20, + medianMs: 15, + maxMs: 200, + p95Ms: 45, + p99Ms: 90, + }, + histogram: [ + { upperBoundMs: 50, count: 960 }, + { upperBoundMs: null, count: 40 }, + ], +} + +const profile: LoadProfile = { + name: 'healthy-api', + targets: [{ name: 'health', endpoint: 'api', path: '/health' }], + vus: 10, + duration: '10s', + thresholds: { + maxP95LatencyMs: 100, + maxP99LatencyMs: 150, + maxErrorRate: 0.01, + minThroughputRps: 50, + }, +} + +class FakeLoadJobClient implements KubernetesLoadJobClient { + applied: Array> = [] + deleted: string[] = [] + waited: string[] = [] + + constructor(private readonly output: LoadMeasurements = measurements) {} + + async apply(resources: Array>): Promise { + this.applied = resources + } + + async waitForCompletion(jobName: string): Promise { + this.waited.push(jobName) + } + + async logs(): Promise { + return `k6 output\n${K6_EVIDENCE_PREFIX}${JSON.stringify(this.output)}\n` + } + + async delete(jobName: string, configMapName: string): Promise { + this.deleted.push(jobName, configMapName) + } +} + +describe('load SLO gate', () => { + it('passes when every measured value is within its declarative threshold', () => { + expect(evaluateLoadSlo(measurements, profile.thresholds)).toEqual({ + passed: true, + violations: [], + }) + }) + + it('fails with every violated metric and measured value', () => { + const result = evaluateLoadSlo(measurements, { + maxP95LatencyMs: 1, + maxP99LatencyMs: 2, + maxErrorRate: 0.001, + minThroughputRps: 200, + }) + + expect(result.passed).toBe(false) + expect(result.violations).toEqual([ + { metric: 'p95LatencyMs', actual: 45, threshold: 1, operator: 'at-most' }, + { metric: 'p99LatencyMs', actual: 90, threshold: 2, operator: 'at-most' }, + { metric: 'errorRate', actual: 0.002, threshold: 0.001, operator: 'at-most' }, + { metric: 'throughputRps', actual: 99.5, threshold: 200, operator: 'at-least' }, + ]) + }) + + it('extracts the machine-readable summary and fails closed on missing evidence', () => { + expect(parseK6LoadMeasurements( + `noise\n${K6_EVIDENCE_PREFIX}${JSON.stringify(measurements)}\n`, + )).toEqual(measurements) + expect(() => parseK6LoadMeasurements('ordinary k6 output')).toThrow(/without Factory evidence JSON/u) + }) +}) + +describe('k6 in-cluster Job', () => { + it('renders weighted request shapes, a ramp, and hardened Job resources', () => { + const resolvedProfile = LoadProfileSchema.parse({ + ...profile, + rps: 50, + maxVus: 100, + ramp: { up: '2s', down: '1s' }, + targets: [{ + name: 'create', + endpoint: 'api', + path: '/items', + method: 'POST', + body: { value: 42 }, + weight: 3, + }], + }) + const resources = createK6LoadJobResources( + { id: 'env-1', endpoints: { api: 'http://api.verify.svc.cluster.local:8080' } }, + resolvedProfile, + { jobName: 'factory-load-test', namespace: 'verify' }, + ) + + expect(k6ScenarioFor(resolvedProfile)).toMatchObject({ + executor: 'ramping-arrival-rate', + stages: [ + { duration: '2s', target: 50 }, + { duration: '10s', target: 50 }, + { duration: '1s', target: 0 }, + ], + }) + expect(resources.resources.map((resource) => resource.kind)).toEqual(['ConfigMap', 'Job']) + expect(JSON.stringify(resources.resources)).toContain('http://api.verify.svc.cluster.local:8080/items') + expect(JSON.stringify(resources.resources)).toContain("'factory_latency_bucket_' + index") + expect(JSON.stringify(resources.resources)).toContain('automountServiceAccountToken') + }) + + it('runs the Job, returns inspectable evidence, and cleans up', async () => { + const client = new FakeLoadJobClient() + const moments = [new Date('2026-07-21T10:00:00.000Z'), new Date('2026-07-21T10:00:10.050Z')] + const result = await runLoad( + { + id: 'env-1', + namespace: 'verify', + endpoints: { api: 'http://api.verify.svc.cluster.local:8080' }, + }, + profile, + { + client, + runId: 'unit-test', + now: () => moments.shift() as Date, + }, + ) + + expect(result).toMatchObject({ + status: 'pass', + passed: true, + measured: measurements, + job: { namespace: 'verify', completed: true }, + evidence: { + contract: 'factory.load.evidence.v1', + environmentId: 'env-1', + gate: { status: 'pass', violations: [] }, + }, + }) + expect(JSON.parse(result.evidenceJson)).toEqual(result.evidence) + expect(client.applied).toHaveLength(2) + expect(client.waited).toHaveLength(1) + expect(client.deleted).toHaveLength(2) + }) + + it('returns a failed gate without treating the completed Job as an execution error', async () => { + const client = new FakeLoadJobClient() + const result = await runLoad( + { id: 'env-1', endpoints: { api: 'http://api:8080' } }, + { + ...profile, + thresholds: { maxP95LatencyMs: 1 }, + }, + { client, runId: 'strict-slo' }, + ) + + expect(result.status).toBe('fail') + expect(result.violations).toContainEqual({ + metric: 'p95LatencyMs', + actual: 45, + threshold: 1, + operator: 'at-most', + }) + }) +}) + +describe('load profile sidecar', () => { + const temporaryDirectories: string[] = [] + + afterEach(async () => { + await Promise.all(temporaryDirectories.splice(0).map(async (path) => await rm(path, { recursive: true }))) + }) + + it('loads YAML and applies safe profile defaults', async () => { + const directory = await mkdtemp(join(tmpdir(), 'factory-load-profile-')) + temporaryDirectories.push(directory) + const path = join(directory, 'load.yaml') + await writeFile(path, ` +name: smoke +vus: 2 +duration: 3s +targets: + - name: health + endpoint: api + path: /health +thresholds: + maxP95LatencyMs: 1000 + maxErrorRate: 0.01 + minThroughputRps: 1 +`, 'utf8') + + const loaded = await loadLoadProfile(path) + expect(loaded.targets[0]).toMatchObject({ method: 'GET', weight: 1 }) + expect(loaded.histogramBucketsMs).toContain(1_000) + }) +}) diff --git a/src/environments/load-harness.ts b/src/environments/load-harness.ts new file mode 100644 index 0000000..5a3dd34 --- /dev/null +++ b/src/environments/load-harness.ts @@ -0,0 +1,272 @@ +import { randomUUID } from 'node:crypto' +import { mkdir, writeFile } from 'node:fs/promises' +import { dirname } from 'node:path' + +import { z } from 'zod' + +import { + createK6LoadJobResources, + K6_EVIDENCE_PREFIX, + KubectlLoadJobClient, + type KubernetesLoadJobClient, + type LoadEnvironment, +} from './k6-job.js' +import { + LoadProfileSchema, + LoadThresholdsSchema, + type LoadProfile, + type ResolvedLoadProfile, + type LoadThresholds, +} from './load-profile.js' + +export const LOAD_EVIDENCE_CONTRACT = 'factory.load.evidence.v1' as const + +const LatencyHistogramBucketSchema = z.object({ + upperBoundMs: z.number().finite().positive().nullable(), + count: z.number().finite().nonnegative(), +}).strict() + +export const LoadMeasurementsSchema = z.object({ + requestCount: z.number().finite().nonnegative(), + errorCount: z.number().finite().nonnegative(), + errorRate: z.number().finite().min(0).max(1), + throughputRps: z.number().finite().nonnegative(), + durationMs: z.number().finite().nonnegative(), + latency: z.object({ + minMs: z.number().finite().nonnegative(), + averageMs: z.number().finite().nonnegative(), + medianMs: z.number().finite().nonnegative(), + maxMs: z.number().finite().nonnegative(), + p95Ms: z.number().finite().nonnegative(), + p99Ms: z.number().finite().nonnegative(), + }).strict(), + histogram: z.array(LatencyHistogramBucketSchema).min(2), +}).strict() + +export type LoadMeasurements = z.infer +export type LatencyHistogramBucket = z.infer + +export type LoadSloMetric = + | 'p95LatencyMs' + | 'p99LatencyMs' + | 'errorRate' + | 'throughputRps' + +export interface LoadSloViolation { + metric: LoadSloMetric + actual: number + threshold: number + operator: 'at-most' | 'at-least' +} + +export interface LoadSloEvaluation { + passed: boolean + violations: LoadSloViolation[] +} + +export interface LoadEvidence { + contract: typeof LOAD_EVIDENCE_CONTRACT + environmentId: string + profile: { + name: string + vus?: number + maxVus?: number + rps?: number + duration: string + ramp?: { up: string; down?: string } + targets: Array<{ + name: string + endpoint?: string + method: string + path?: string + }> + } + job: { + name: string + namespace: string + completed: true + } + startedAt: string + completedAt: string + thresholds: LoadThresholds + measured: LoadMeasurements + gate: { + status: 'pass' | 'fail' + violations: LoadSloViolation[] + } +} + +export interface LoadResult { + status: 'pass' | 'fail' + passed: boolean + measured: LoadMeasurements + violations: LoadSloViolation[] + evidence: LoadEvidence + evidenceJson: string + job: LoadEvidence['job'] +} + +export interface RunLoadOptions { + client?: KubernetesLoadJobClient + namespace?: string + kubeContext?: string + k6Image?: string + timeoutMs?: number + cleanup?: boolean + evidencePath?: string + runId?: string + now?: () => Date +} + +export function evaluateLoadSlo( + measuredInput: LoadMeasurements, + thresholdsInput: LoadThresholds, +): LoadSloEvaluation { + const measured = LoadMeasurementsSchema.parse(measuredInput) + const thresholds = LoadThresholdsSchema.parse(thresholdsInput) + const violations: LoadSloViolation[] = [] + + const atMost = (metric: LoadSloMetric, actual: number, threshold: number | undefined): void => { + if (threshold !== undefined && actual > threshold) { + violations.push({ metric, actual, threshold, operator: 'at-most' }) + } + } + const atLeast = (metric: LoadSloMetric, actual: number, threshold: number | undefined): void => { + if (threshold !== undefined && actual < threshold) { + violations.push({ metric, actual, threshold, operator: 'at-least' }) + } + } + + atMost('p95LatencyMs', measured.latency.p95Ms, thresholds.maxP95LatencyMs) + atMost('p99LatencyMs', measured.latency.p99Ms, thresholds.maxP99LatencyMs) + atMost('errorRate', measured.errorRate, thresholds.maxErrorRate) + atLeast('throughputRps', measured.throughputRps, thresholds.minThroughputRps) + return { passed: violations.length === 0, violations } +} + +export function parseK6LoadMeasurements(logs: string): LoadMeasurements { + const markerIndex = logs.lastIndexOf(K6_EVIDENCE_PREFIX) + if (markerIndex < 0) { + throw new Error('k6 Job completed without Factory evidence JSON') + } + const jsonLine = logs + .slice(markerIndex + K6_EVIDENCE_PREFIX.length) + .split(/\r?\n/u, 1)[0] + .trim() + let value: unknown + try { + value = JSON.parse(jsonLine) + } catch (error) { + throw new Error(`k6 Job emitted invalid Factory evidence JSON: ${error instanceof Error ? error.message : String(error)}`) + } + const parsed = LoadMeasurementsSchema.safeParse(value) + if (!parsed.success) { + const issues = parsed.error.issues + .map((issue) => `${issue.path.join('.') || ''}: ${issue.message}`) + .join('; ') + throw new Error(`k6 Job evidence did not match the expected schema: ${issues}`) + } + return parsed.data +} + +export function serializeLoadEvidence(evidence: LoadEvidence): string { + return `${JSON.stringify(evidence, null, 2)}\n` +} + +const evidenceProfile = (profile: ResolvedLoadProfile): LoadEvidence['profile'] => ({ + name: profile.name, + ...(profile.vus === undefined ? {} : { vus: profile.vus }), + ...(profile.maxVus === undefined ? {} : { maxVus: profile.maxVus }), + ...(profile.rps === undefined ? {} : { rps: profile.rps }), + duration: profile.duration, + ...(profile.ramp === undefined ? {} : { ramp: profile.ramp }), + targets: profile.targets.map((target) => ({ + name: target.name, + ...(target.endpoint === undefined ? {} : { endpoint: target.endpoint }), + method: target.method, + ...(target.path === undefined ? {} : { path: target.path }), + })), +}) + +export async function runLoad( + environment: LoadEnvironment, + profileInput: LoadProfile, + options: RunLoadOptions = {}, +): Promise { + const profile = LoadProfileSchema.parse(profileInput) + const namespace = options.namespace ?? environment.namespace ?? 'default' + const requestedRunId = (options.runId ?? randomUUID()) + .replace(/[^a-zA-Z0-9]+/gu, '-') + .replace(/^-+|-+$/gu, '') + .slice(0, 12) + const runId = requestedRunId || randomUUID().replaceAll('-', '').slice(0, 12) + const jobResources = createK6LoadJobResources(environment, profile, { + jobName: `factory-load-${profile.name.slice(0, 14)}-${runId}`, + namespace, + image: options.k6Image, + timeoutMs: options.timeoutMs, + }) + const client = options.client ?? new KubectlLoadJobClient({ + context: options.kubeContext ?? environment.kubeContext, + }) + const now = options.now ?? (() => new Date()) + const startedAt = now().toISOString() + let mainError: unknown + + try { + await client.apply(jobResources.resources, namespace) + await client.waitForCompletion(jobResources.jobName, namespace, jobResources.timeoutMs) + const logs = await client.logs(jobResources.jobName, namespace) + const measured = parseK6LoadMeasurements(logs) + const gate = evaluateLoadSlo(measured, profile.thresholds) + const status = gate.passed ? 'pass' : 'fail' + const job: LoadEvidence['job'] = { + name: jobResources.jobName, + namespace, + completed: true, + } + const evidence: LoadEvidence = { + contract: LOAD_EVIDENCE_CONTRACT, + environmentId: environment.id, + profile: evidenceProfile(profile), + job, + startedAt, + completedAt: now().toISOString(), + thresholds: profile.thresholds, + measured, + gate: { status, violations: gate.violations }, + } + const evidenceJson = serializeLoadEvidence(evidence) + if (options.evidencePath) { + await mkdir(dirname(options.evidencePath), { recursive: true }) + await writeFile(options.evidencePath, evidenceJson, 'utf8') + } + return { + status, + passed: gate.passed, + measured, + violations: gate.violations, + evidence, + evidenceJson, + job, + } + } catch (error) { + mainError = error + throw error + } finally { + if (options.cleanup !== false) { + try { + await client.delete(jobResources.jobName, jobResources.configMapName, namespace) + } catch (cleanupError) { + if (mainError === undefined) throw cleanupError + } + } + } +} + +export type { + KubernetesLoadJobClient, + LoadEnvironment, + LoadProfile, + LoadThresholds, +} diff --git a/src/environments/load-profile.ts b/src/environments/load-profile.ts new file mode 100644 index 0000000..5f8e27e --- /dev/null +++ b/src/environments/load-profile.ts @@ -0,0 +1,143 @@ +import { readFile } from 'node:fs/promises' + +import { parse as parseYaml } from 'yaml' +import { z } from 'zod' + +const duration = z.string().trim().regex( + /^\d+(?:\.\d+)?(?:ms|s|m|h)$/u, + 'must be a k6 duration such as 500ms, 30s, 5m, or 1h', +).refine((value) => !/^0+(?:\.0+)?(?:ms|s|m|h)$/u.test(value), 'must be greater than zero') + +const positiveFinite = z.number().finite().positive() +const nonNegativeFinite = z.number().finite().nonnegative() + +export const LoadThresholdsSchema = z.object({ + maxP95LatencyMs: positiveFinite.optional(), + maxP99LatencyMs: positiveFinite.optional(), + maxErrorRate: z.number().finite().min(0).max(1).optional(), + minThroughputRps: nonNegativeFinite.optional(), +}).strict().refine( + (thresholds) => Object.values(thresholds).some((value) => value !== undefined), + { message: 'at least one SLO threshold is required' }, +) + +export type LoadThresholds = z.infer + +export const LoadTargetSchema = z.object({ + name: z.string().trim().min(1).max(80), + endpoint: z.string().trim().min(1).optional(), + url: z.string().url().optional(), + path: z.string().optional(), + method: z.enum(['GET', 'POST', 'PUT', 'PATCH', 'DELETE', 'HEAD', 'OPTIONS']).default('GET'), + headers: z.record(z.string()).optional(), + body: z.unknown().optional(), + expectedStatuses: z.array(z.number().int().min(100).max(599)).min(1).optional(), + weight: z.number().int().positive().max(10_000).default(1), +}).strict().superRefine((target, context) => { + if ((target.endpoint === undefined) === (target.url === undefined)) { + context.addIssue({ + code: z.ZodIssueCode.custom, + path: ['endpoint'], + message: 'set exactly one of endpoint or url', + }) + } + if (target.url !== undefined && !/^https?:\/\//u.test(target.url)) { + context.addIssue({ + code: z.ZodIssueCode.custom, + path: ['url'], + message: 'only http and https targets are supported', + }) + } +}) + +export type LoadTarget = z.input +export type ResolvedLoadTargetProfile = z.output + +export const LoadProfileSchema = z.object({ + name: z.string().trim().min(1).max(80), + targets: z.array(LoadTargetSchema).min(1), + vus: z.number().int().positive().max(100_000).optional(), + maxVus: z.number().int().positive().max(100_000).optional(), + rps: z.number().int().positive().max(1_000_000).optional(), + duration, + ramp: z.object({ + up: duration, + down: duration.optional(), + }).strict().optional(), + thresholds: LoadThresholdsSchema, + histogramBucketsMs: z.array(positiveFinite).min(1).max(50) + .default([1, 5, 10, 25, 50, 100, 250, 500, 1_000, 2_000, 5_000]), +}).strict().superRefine((profile, context) => { + if (profile.vus === undefined && profile.rps === undefined) { + context.addIssue({ + code: z.ZodIssueCode.custom, + path: ['vus'], + message: 'set vus, rps, or both', + }) + } + if (profile.maxVus !== undefined && profile.rps === undefined) { + context.addIssue({ + code: z.ZodIssueCode.custom, + path: ['maxVus'], + message: 'maxVus is only used with an rps profile', + }) + } + if (profile.vus !== undefined && profile.maxVus !== undefined && profile.maxVus < profile.vus) { + context.addIssue({ + code: z.ZodIssueCode.custom, + path: ['maxVus'], + message: 'maxVus must be greater than or equal to vus', + }) + } + const sortedBuckets = [...profile.histogramBucketsMs].sort((left, right) => left - right) + if (sortedBuckets.some((bucket, index) => bucket !== profile.histogramBucketsMs[index])) { + context.addIssue({ + code: z.ZodIssueCode.custom, + path: ['histogramBucketsMs'], + message: 'histogram buckets must be in ascending order', + }) + } + if (new Set(profile.histogramBucketsMs).size !== profile.histogramBucketsMs.length) { + context.addIssue({ + code: z.ZodIssueCode.custom, + path: ['histogramBucketsMs'], + message: 'histogram buckets must be unique', + }) + } +}) + +/** Declarative profile input. Schema defaults make method, weight, and histogram buckets optional. */ +export type LoadProfile = z.input +export type ResolvedLoadProfile = z.output + +export async function loadLoadProfile(path: string): Promise { + const source = await readFile(path, 'utf8') + let value: unknown + try { + value = parseYaml(source) + } catch (error) { + throw new Error(`Could not parse load profile ${path}: ${error instanceof Error ? error.message : String(error)}`) + } + + const parsed = LoadProfileSchema.safeParse(value) + if (!parsed.success) { + const issues = parsed.error.issues + .map((issue) => `${issue.path.join('.') || ''}: ${issue.message}`) + .join('; ') + throw new Error(`Invalid load profile ${path}: ${issues}`) + } + return parsed.data +} + +export function durationToMilliseconds(value: string): number { + const match = /^(\d+(?:\.\d+)?)(ms|s|m|h)$/u.exec(value) + if (!match) throw new Error(`Invalid duration: ${value}`) + const amount = Number(match[1]) + const multiplier = { + ms: 1, + s: 1_000, + m: 60_000, + h: 3_600_000, + }[match[2] as 'ms' | 's' | 'm' | 'h'] + return amount * multiplier +} diff --git a/src/index.ts b/src/index.ts index 2f25b13..192a827 100644 --- a/src/index.ts +++ b/src/index.ts @@ -358,3 +358,55 @@ export type { TriageDecision, TriageEngine, } from './types' +export { + LOAD_EVIDENCE_CONTRACT, + LoadMeasurementsSchema, + evaluateLoadSlo, + parseK6LoadMeasurements, + runLoad, + serializeLoadEvidence, +} from './environments/load-harness' +export type { + KubernetesLoadJobClient, + LatencyHistogramBucket, + LoadEnvironment, + LoadEvidence, + LoadMeasurements, + LoadProfile, + LoadResult, + LoadSloEvaluation, + LoadSloMetric, + LoadSloViolation, + LoadThresholds, + RunLoadOptions, +} from './environments/load-harness' +export { + LoadProfileSchema, + LoadTargetSchema, + LoadThresholdsSchema, + durationToMilliseconds, + loadLoadProfile, +} from './environments/load-profile' +export type { + LoadTarget, + ResolvedLoadProfile, + ResolvedLoadTargetProfile, +} from './environments/load-profile' +export { + DEFAULT_K6_IMAGE, + K6_EVIDENCE_PREFIX, + KubectlLoadJobClient, + createK6LoadJobResources, + defaultKubectlCommandRunner, + k6ScenarioFor, + renderK6Script, + resolveLoadTargets, +} from './environments/k6-job' +export type { + CreateK6LoadJobOptions, + K6LoadJobResources, + KubectlCommandResult, + KubectlCommandRunner, + KubectlLoadJobClientOptions, + ResolvedLoadTarget, +} from './environments/k6-job' diff --git a/test/e2e/load-harness.e2e.ts b/test/e2e/load-harness.e2e.ts new file mode 100644 index 0000000..52a33c1 --- /dev/null +++ b/test/e2e/load-harness.e2e.ts @@ -0,0 +1,151 @@ +import assert from 'node:assert/strict' +import { mkdir } from 'node:fs/promises' + +import { + defaultKubectlCommandRunner, + runLoad, + type LoadEnvironment, + type LoadProfile, +} from '../../src/index.ts' + +const kubectl = defaultKubectlCommandRunner() +const namespace = `factory-load-e2e-${process.pid}` +const evidenceDirectory = 'artifacts/load-e2e' + +const apply = async (resources: Array>): Promise => { + await kubectl( + ['--namespace', namespace, 'apply', '--filename', '-'], + JSON.stringify({ apiVersion: 'v1', kind: 'List', items: resources }), + ) +} + +const provisionEnvironment = async (): Promise => { + await kubectl(['create', 'namespace', namespace]) + return { + id: namespace, + namespace, + endpoints: { + api: `http://sample-api.${namespace}.svc.cluster.local:5678`, + }, + } +} + +const deploySampleService = async (): Promise => { + const labels = { app: 'factory-load-sample' } + await apply([ + { + apiVersion: 'apps/v1', + kind: 'Deployment', + metadata: { name: 'sample-api', namespace }, + spec: { + replicas: 2, + selector: { matchLabels: labels }, + template: { + metadata: { labels }, + spec: { + containers: [{ + name: 'http-echo', + image: 'hashicorp/http-echo:1.0.0@sha256:fcb75f691c8b0414d670ae570240cbf95502cc18a9ba57e982ecac589760a186', + args: ['-listen=:5678', '-text={"status":"healthy"}'], + ports: [{ name: 'http', containerPort: 5678 }], + readinessProbe: { + httpGet: { path: '/', port: 'http' }, + initialDelaySeconds: 1, + periodSeconds: 1, + }, + resources: { + requests: { cpu: '20m', memory: '16Mi' }, + limits: { cpu: '250m', memory: '64Mi' }, + }, + }], + }, + }, + }, + }, + { + apiVersion: 'v1', + kind: 'Service', + metadata: { name: 'sample-api', namespace }, + spec: { + selector: labels, + ports: [{ name: 'http', port: 5678, targetPort: 'http' }], + }, + }, + ]) + await kubectl([ + '--namespace', namespace, + 'wait', 'deployment/sample-api', + '--for=condition=Available', + '--timeout=120s', + ]) +} + +const passingProfile = (): LoadProfile => ({ + name: 'healthy-service-pass', + targets: [{ name: 'health', endpoint: 'api', path: '/' }], + vus: Number(process.env.FACTORY_LOAD_E2E_VUS ?? 50), + duration: process.env.FACTORY_LOAD_E2E_DURATION ?? '30s', + ramp: { up: '2s', down: '1s' }, + thresholds: { + maxP95LatencyMs: 2_000, + maxP99LatencyMs: 5_000, + maxErrorRate: 0.01, + minThroughputRps: 1, + }, +}) + +const strictProfile = (): LoadProfile => ({ + name: 'healthy-service-strict-fail', + targets: [{ name: 'health', endpoint: 'api', path: '/' }], + vus: 5, + duration: '3s', + thresholds: { + maxP95LatencyMs: 0.001, + maxErrorRate: 0.01, + minThroughputRps: 1, + }, +}) + +const main = async (): Promise => { + await mkdir(evidenceDirectory, { recursive: true }) + let environment: LoadEnvironment | undefined + try { + environment = await provisionEnvironment() + await deploySampleService() + + const passing = await runLoad(environment, passingProfile(), { + cleanup: false, + evidencePath: `${evidenceDirectory}/pass.json`, + }) + assert.equal(passing.status, 'pass', passing.evidenceJson) + assert.equal(passing.job.completed, true) + assert.ok(passing.measured.requestCount > 0, 'pass run must emit a real request count') + assert.ok(passing.measured.throughputRps > 0, 'pass run must emit real throughput') + assert.ok(passing.measured.latency.p95Ms > 0, 'pass run must emit a real p95') + assert.ok(passing.measured.histogram.some((bucket) => bucket.count > 0), 'latency histogram must not be empty') + assert.deepEqual(JSON.parse(passing.evidenceJson), passing.evidence) + + const failing = await runLoad(environment, strictProfile(), { + cleanup: false, + evidencePath: `${evidenceDirectory}/fail.json`, + }) + assert.equal(failing.status, 'fail', 'strict SLO must reject a completed load run') + assert.equal(failing.job.completed, true) + assert.ok(failing.violations.some((violation) => violation.metric === 'p95LatencyMs')) + assert.ok(failing.measured.requestCount > 0, 'failed gate must retain measured evidence') + + process.stdout.write(`${JSON.stringify({ + passing: passing.evidence, + failing: failing.evidence, + }, null, 2)}\n`) + } finally { + if (environment) { + await kubectl(['delete', 'namespace', namespace, '--wait=false', '--ignore-not-found=true']) + } + } +} + +main().catch((error: unknown) => { + process.stderr.write(`${error instanceof Error ? error.stack : String(error)}\n`) + process.exitCode = 1 +}) From 17718126137c477d3436790736f5a317bed8c861 Mon Sep 17 00:00:00 2001 From: kjgbot Date: Tue, 21 Jul 2026 09:56:51 +0200 Subject: [PATCH 2/3] harden load evidence gates --- .github/workflows/ci.yml | 4 + docs/load-harness.md | 2 + src/environments/load-harness.test.ts | 43 +++++++++- src/environments/load-harness.ts | 105 +++++++++++++++++++++++- src/environments/load-profile.ts | 7 +- src/fleet/internal-fleet-client.test.ts | 8 +- test/e2e/load-harness.e2e.ts | 2 +- 7 files changed, 162 insertions(+), 9 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 2cc8c8b..323edb5 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -65,6 +65,10 @@ jobs: steps: - name: Checkout uses: actions/checkout@v6 + with: + # Keep the load evidence bound to the reviewed head, matching the + # package job's head-bound attestation. + ref: ${{ github.event.pull_request.head.sha || github.sha }} - name: Setup Node uses: actions/setup-node@v6 diff --git a/docs/load-harness.md b/docs/load-harness.md index 093b628..ef20867 100644 --- a/docs/load-harness.md +++ b/docs/load-harness.md @@ -55,6 +55,8 @@ error rate, throughput, a cumulative latency histogram, thresholds, and every violated metric. Request headers and bodies are deliberately omitted from evidence. The ephemeral ConfigMap and Job are deleted by default; pass `cleanup: false` when a surrounding environment teardown owns cleanup. +The gate also fails closed with a `requestCount` violation when k6 completes +without sending any requests, even if the profile omitted a throughput floor. Run the real kind-cluster proof with: diff --git a/src/environments/load-harness.test.ts b/src/environments/load-harness.test.ts index 4720aa0..9c7b00d 100644 --- a/src/environments/load-harness.test.ts +++ b/src/environments/load-harness.test.ts @@ -100,11 +100,43 @@ describe('load SLO gate', () => { ]) }) - it('extracts the machine-readable summary and fails closed on missing evidence', () => { + it('fails closed when a completed generator emitted no requests', () => { + const empty: LoadMeasurements = { + requestCount: 0, + errorCount: 0, + errorRate: 1, + throughputRps: 0, + durationMs: 1_000, + latency: { + minMs: 0, + averageMs: 0, + medianMs: 0, + maxMs: 0, + p95Ms: 0, + p99Ms: 0, + }, + histogram: [ + { upperBoundMs: 50, count: 0 }, + { upperBoundMs: null, count: 0 }, + ], + } + + expect(evaluateLoadSlo(empty, { maxP95LatencyMs: 2_000 })).toEqual({ + passed: false, + violations: [ + { metric: 'requestCount', actual: 0, threshold: 1, operator: 'at-least' }, + ], + }) + }) + + it('extracts valid summaries and rejects missing or internally inconsistent evidence', () => { expect(parseK6LoadMeasurements( `noise\n${K6_EVIDENCE_PREFIX}${JSON.stringify(measurements)}\n`, )).toEqual(measurements) expect(() => parseK6LoadMeasurements('ordinary k6 output')).toThrow(/without Factory evidence JSON/u) + expect(() => parseK6LoadMeasurements( + `${K6_EVIDENCE_PREFIX}${JSON.stringify({ ...measurements, errorCount: 1_001 })}\n`, + )).toThrow(/errorCount: must not exceed requestCount/u) }) }) @@ -228,4 +260,13 @@ thresholds: expect(loaded.targets[0]).toMatchObject({ method: 'GET', weight: 1 }) expect(loaded.histogramBucketsMs).toContain(1_000) }) + + it('rejects an arrival-rate profile whose maxVus is below its default preallocation', () => { + expect(() => LoadProfileSchema.parse({ + ...profile, + vus: undefined, + rps: 100, + maxVus: 50, + })).toThrow(/preallocated vus/u) + }) }) diff --git a/src/environments/load-harness.ts b/src/environments/load-harness.ts index 5a3dd34..e4d2822 100644 --- a/src/environments/load-harness.ts +++ b/src/environments/load-harness.ts @@ -23,12 +23,12 @@ export const LOAD_EVIDENCE_CONTRACT = 'factory.load.evidence.v1' as const const LatencyHistogramBucketSchema = z.object({ upperBoundMs: z.number().finite().positive().nullable(), - count: z.number().finite().nonnegative(), + count: z.number().int().nonnegative(), }).strict() export const LoadMeasurementsSchema = z.object({ - requestCount: z.number().finite().nonnegative(), - errorCount: z.number().finite().nonnegative(), + requestCount: z.number().int().nonnegative(), + errorCount: z.number().int().nonnegative(), errorRate: z.number().finite().min(0).max(1), throughputRps: z.number().finite().nonnegative(), durationMs: z.number().finite().nonnegative(), @@ -41,12 +41,106 @@ export const LoadMeasurementsSchema = z.object({ p99Ms: z.number().finite().nonnegative(), }).strict(), histogram: z.array(LatencyHistogramBucketSchema).min(2), -}).strict() +}).strict().superRefine((measured, context) => { + if (measured.errorCount > measured.requestCount) { + context.addIssue({ + code: z.ZodIssueCode.custom, + path: ['errorCount'], + message: 'must not exceed requestCount', + }) + } + + const expectedErrorRate = measured.requestCount === 0 + ? 1 + : measured.errorCount / measured.requestCount + if (Math.abs(measured.errorRate - expectedErrorRate) > 1e-12) { + context.addIssue({ + code: z.ZodIssueCode.custom, + path: ['errorRate'], + message: 'must equal errorCount / requestCount (or 1 when no requests ran)', + }) + } + + const latencyOrder = [ + measured.latency.minMs, + measured.latency.medianMs, + measured.latency.p95Ms, + measured.latency.p99Ms, + measured.latency.maxMs, + ] + if (latencyOrder.some((value, index) => index > 0 && value < latencyOrder[index - 1])) { + context.addIssue({ + code: z.ZodIssueCode.custom, + path: ['latency'], + message: 'must satisfy min <= median <= p95 <= p99 <= max', + }) + } + if (measured.latency.averageMs < measured.latency.minMs || + measured.latency.averageMs > measured.latency.maxMs) { + context.addIssue({ + code: z.ZodIssueCode.custom, + path: ['latency', 'averageMs'], + message: 'must be between minMs and maxMs', + }) + } + + const finalBucketIndex = measured.histogram.length - 1 + measured.histogram.forEach((bucket, index) => { + const previous = measured.histogram[index - 1] + if (index === finalBucketIndex && bucket.upperBoundMs !== null) { + context.addIssue({ + code: z.ZodIssueCode.custom, + path: ['histogram', index, 'upperBoundMs'], + message: 'the final overflow bucket must have a null upper bound', + }) + } else if (index < finalBucketIndex && bucket.upperBoundMs === null) { + context.addIssue({ + code: z.ZodIssueCode.custom, + path: ['histogram', index, 'upperBoundMs'], + message: 'only the final overflow bucket may have a null upper bound', + }) + } + if (bucket.count > measured.requestCount) { + context.addIssue({ + code: z.ZodIssueCode.custom, + path: ['histogram', index, 'count'], + message: 'must not exceed requestCount', + }) + } + if (previous !== undefined && previous.upperBoundMs !== null && bucket.upperBoundMs !== null) { + if (bucket.upperBoundMs <= previous.upperBoundMs) { + context.addIssue({ + code: z.ZodIssueCode.custom, + path: ['histogram', index, 'upperBoundMs'], + message: 'finite upper bounds must be strictly increasing', + }) + } + if (bucket.count < previous.count) { + context.addIssue({ + code: z.ZodIssueCode.custom, + path: ['histogram', index, 'count'], + message: 'cumulative bucket counts must be nondecreasing', + }) + } + } + }) + + const lastFiniteCount = measured.histogram[finalBucketIndex - 1]?.count ?? 0 + const overflowCount = measured.histogram[finalBucketIndex]?.count ?? 0 + if (lastFiniteCount + overflowCount !== measured.requestCount) { + context.addIssue({ + code: z.ZodIssueCode.custom, + path: ['histogram'], + message: 'last finite bucket plus overflow bucket must equal requestCount', + }) + } +}) export type LoadMeasurements = z.infer export type LatencyHistogramBucket = z.infer export type LoadSloMetric = + | 'requestCount' | 'p95LatencyMs' | 'p99LatencyMs' | 'errorRate' @@ -137,6 +231,9 @@ export function evaluateLoadSlo( } } + // A completed generator with no traffic is never evidence that an SLO passed, + // even when the caller did not declare a throughput floor. + atLeast('requestCount', measured.requestCount, 1) atMost('p95LatencyMs', measured.latency.p95Ms, thresholds.maxP95LatencyMs) atMost('p99LatencyMs', measured.latency.p99Ms, thresholds.maxP99LatencyMs) atMost('errorRate', measured.errorRate, thresholds.maxErrorRate) diff --git a/src/environments/load-profile.ts b/src/environments/load-profile.ts index 5f8e27e..2256902 100644 --- a/src/environments/load-profile.ts +++ b/src/environments/load-profile.ts @@ -82,11 +82,14 @@ export const LoadProfileSchema = z.object({ message: 'maxVus is only used with an rps profile', }) } - if (profile.vus !== undefined && profile.maxVus !== undefined && profile.maxVus < profile.vus) { + const preAllocatedVus = profile.rps === undefined + ? undefined + : (profile.vus ?? profile.rps) + if (profile.maxVus !== undefined && preAllocatedVus !== undefined && profile.maxVus < preAllocatedVus) { context.addIssue({ code: z.ZodIssueCode.custom, path: ['maxVus'], - message: 'maxVus must be greater than or equal to vus', + message: 'maxVus must be greater than or equal to the preallocated vus', }) } const sortedBuckets = [...profile.histogramBucketsMs].sort((left, right) => left - right) diff --git a/src/fleet/internal-fleet-client.test.ts b/src/fleet/internal-fleet-client.test.ts index 2e4a026..475be3a 100644 --- a/src/fleet/internal-fleet-client.test.ts +++ b/src/fleet/internal-fleet-client.test.ts @@ -704,7 +704,13 @@ describe('InternalFleetClient', () => { it('surfaces the broker pid as protected process state', async () => { const harness = new FakeHarnessDriverClient() harness.brokerPid = 68009 - const fleet = new InternalFleetClient({ client: harness, cwd: '/worktree' }) + const fleet = new InternalFleetClient({ + client: harness, + cwd: '/worktree', + // Keep the unit isolated from an AGENT_RELAY_STATE_DIR inherited by the + // test process; this assertion is specifically about the injected client. + connectionPath: 'test/fixtures/no-such-broker-connection.json', + }) await expect(fleet.protectedPids()).resolves.toEqual([68009]) }) diff --git a/test/e2e/load-harness.e2e.ts b/test/e2e/load-harness.e2e.ts index 52a33c1..f17f3a2 100644 --- a/test/e2e/load-harness.e2e.ts +++ b/test/e2e/load-harness.e2e.ts @@ -140,7 +140,7 @@ const main = async (): Promise => { }, null, 2)}\n`) } finally { if (environment) { - await kubectl(['delete', 'namespace', namespace, '--wait=false', '--ignore-not-found=true']) + await kubectl(['delete', 'namespace', namespace, '--wait=true', '--ignore-not-found=true']) } } } From 94d24549c3dcc744027076044ea8dff903914867 Mon Sep 17 00:00:00 2001 From: kjgbot Date: Tue, 21 Jul 2026 12:24:11 +0200 Subject: [PATCH 3/3] test: pin load e2e kubernetes context --- test/e2e/load-harness.e2e.ts | 15 ++++++++++++++- 1 file changed, 14 insertions(+), 1 deletion(-) diff --git a/test/e2e/load-harness.e2e.ts b/test/e2e/load-harness.e2e.ts index f17f3a2..0c9fc2d 100644 --- a/test/e2e/load-harness.e2e.ts +++ b/test/e2e/load-harness.e2e.ts @@ -8,7 +8,12 @@ import { type LoadProfile, } from '../../src/index.ts' -const kubectl = defaultKubectlCommandRunner() +const rawKubectl = defaultKubectlCommandRunner() +let kubeContext = '' +const kubectl = async (args: string[], input?: string) => await rawKubectl( + kubeContext ? ['--context', kubeContext, ...args] : args, + input, +) const namespace = `factory-load-e2e-${process.pid}` const evidenceDirectory = 'artifacts/load-e2e' @@ -21,9 +26,15 @@ const apply = async (resources: Array>): Promise = const provisionEnvironment = async (): Promise => { await kubectl(['create', 'namespace', namespace]) + await kubectl([ + 'wait', `namespace/${namespace}`, + '--for=jsonpath={.status.phase}=Active', + '--timeout=30s', + ]) return { id: namespace, namespace, + kubeContext, endpoints: { api: `http://sample-api.${namespace}.svc.cluster.local:5678`, }, @@ -108,6 +119,8 @@ const strictProfile = (): LoadProfile => ({ const main = async (): Promise => { await mkdir(evidenceDirectory, { recursive: true }) + kubeContext = (await rawKubectl(['config', 'current-context'])).stdout.trim() + assert.ok(kubeContext, 'load E2E requires a current Kubernetes context') let environment: LoadEnvironment | undefined try { environment = await provisionEnvironment()