#!/usr/bin/env python3 """Interleaved before/after/nginx process benchmarks with immutable raw evidence. This is a same-host diagnostic experiment, not a simulation or release sign-off. Runs >=5 measured samples per variant/scenario by default, each >=15 seconds. """ from __future__ import annotations import argparse import csv import hashlib import itertools import json import os from pathlib import Path import platform import resource import statistics import subprocess import sys import time import traceback from common import Fixture, http, seal_results, sha256, write_json from resources import (Monitor, affinity_snapshot, cgroup_cpu_delta, cgroup_cpu_stat, parse_cpu_set, validate_cpu_sets, wait_for_affinity) from telemetry import traffic_analytics_delta, traffic_analytics_snapshot SIZES = {"small": "small.html", "large": "large.bin", "huge": "huge.bin"} DROP_COUNTER_FIELDS = { "logs_dropped": "history_dropped_delta", "logging.sinks.queue_dropped": "logging_sinks_queue_dropped_delta", "logging.error_file.queue_dropped": "logging_error_file_queue_dropped_delta", "cache.cold_queue_dropped": "cache_cold_queue_dropped_delta", } DROP_INTERVAL_NOTE = ("Delta brackets measured wrk plus one postflight request and possible in-flight completions; " "percentage uses wrk completed requests as an approximate denominator. " "History rejection does not imply rejection by live broadcast or external sinks.") def usage_snapshot(): usage = resource.getrusage(resource.RUSAGE_CHILDREN) return usage.ru_utime + usage.ru_stime def effective_threads(requested, concurrency): if requested < 1 or concurrency < 1: raise ValueError("client threads and concurrency must be positive") # wrk floor-divides the connection count between threads. Select a divisor # so its advertised and effective connection counts remain identical. return next(value for value in range(min(requested, concurrency), 0, -1) if concurrency % value == 0) def wrk_run(args, port, path, duration, concurrency, prefix, roots=None, expected_processes=None): threads = effective_threads(args.client_threads, concurrency) command = [str(args.wrk), "--latency", "-t", str(threads), "-c", str(concurrency), "-d", f"{duration}s", "--timeout", "5s", "-H", "Accept-Encoding: identity", "-s", str(Path(__file__).with_name("wrk_report.lua")), f"http://127.0.0.1:{port}/{path}"] if args.client_cpus: command = ["taskset", "-c", args.client_cpus, *command] affinities = {} expected_processes = expected_processes or {} if roots: for label, pid in roots.items(): expected = parse_cpu_set(args.server_cpus if label == "server" else args.origin_cpus) affinities[label] = affinity_snapshot(pid, expected, expected_processes.get(label)) monitor = Monitor(roots, args.resource_interval, expected_processes) if roots else None if monitor: monitor.start() cpu_start = usage_snapshot() started = time.monotonic() process = None stdout, stderr = "", "" try: process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) affinities["generator"] = wait_for_affinity(process, parse_cpu_set(args.client_cpus)) stdout, stderr = process.communicate(timeout=duration + 30) finally: if process is not None and process.poll() is None: process.kill() stdout, stderr = process.communicate(timeout=5) if monitor: monitor.stop() prefix.parent.mkdir(parents=True, exist_ok=True) prefix.with_suffix(".txt").write_text(stdout + "\n" + stderr) write_json(prefix.with_name(prefix.name + "-affinity.json"), affinities) if monitor: write_json(prefix.with_name(prefix.name + "-resources.json"), monitor.samples) if monitor.errors: write_json(prefix.with_name(prefix.name + "-resource-errors.json"), monitor.errors) cpu_seconds = usage_snapshot() - cpu_start wall_seconds = time.monotonic() - started raw = stdout + "\n" + stderr if process.returncode: raise RuntimeError(f"wrk exited {process.returncode}: {raw[-2000:]}") line = next((line for line in stdout.splitlines() if line.startswith("RUSTPROXY_BENCH_JSON ")), None) if line is None: raise RuntimeError("wrk produced no structured report") result = json.loads(line.removeprefix("RUSTPROXY_BENCH_JSON ")) if result["requests"] <= 0 or result["duration_us"] <= 0: write_json(prefix.with_suffix(".json"), result) raise RuntimeError("wrk completed no requests or reported an invalid duration; raw evidence retained") result.update({"command": command, "wall_seconds": wall_seconds, "client_threads": threads, "requested_connections": concurrency, "client_threads_requested": args.client_threads, "effective_connections": threads * (concurrency // threads), "generator_cpu_seconds": cpu_seconds, "rps": result["requests"] * 1_000_000 / result["duration_us"], "error_total": sum(value for key, value in result.items() if key.startswith("errors_"))}) if monitor: result["resources"] = monitor.summary(result["requests"]) write_json(prefix.with_suffix(".json"), result) return result def payload_check(port, path, expected): """A full exact-body check outside the wrk measurement interval.""" try: status, headers, body = http(port, "/" + path, headers={"Accept-Encoding": "identity"}) return {"status": status, "bytes": len(body), "sha256": hashlib.sha256(body).hexdigest(), "expected_bytes": len(expected), "expected_sha256": hashlib.sha256(expected).hexdigest(), "content_length": headers.get("content-length"), "content_encoding": headers.get("content-encoding"), "approximate_header_bytes": 19 + sum(len(key) + len(value) + 4 for key, value in headers.items()), "passed": status == 200 and body == expected} except Exception as error: return {"passed": False, "error": str(error), "expected_bytes": len(expected), "expected_sha256": hashlib.sha256(expected).hexdigest()} def transfer_sanity(result, payload_bytes, header_bytes, concurrency): """Check gross truncation/size anomalies; this is not per-response hashing.""" requests, wire_bytes = result["requests"], result["bytes"] minimum = requests * payload_bytes # wrk counts headers and partially received final responses in its wire bytes. header_allowance = max(1024, 2 * header_bytes) maximum = (requests + concurrency) * (payload_bytes + header_allowance) return {"passed": requests > 0 and minimum <= wire_bytes <= maximum, "wire_bytes_per_request": wire_bytes / requests if requests else None, "expected_payload_bytes_per_request": payload_bytes, "minimum_wire_bytes": minimum, "maximum_wire_bytes": maximum, "header_allowance_per_response": header_allowance, "inflight_response_allowance": concurrency} def counter_values(value, prefix=""): """Retain numeric drop and rotation counters without session/config data.""" found = {} if isinstance(value, dict): for key, child in value.items(): path = f"{prefix}.{key}" if prefix else key if isinstance(child, (int, float)) and any(part in key for part in ["dropped", "rotations", "generation", "requests_in_generation"]): found[path] = child found.update(counter_values(child, path)) elif isinstance(value, list): for index, child in enumerate(value): found.update(counter_values(child, f"{prefix}[{index}]")) return found def validate_drop_counters(values): for name in DROP_COUNTER_FIELDS: if type(values.get(name)) is not int or values[name] < 0: raise RuntimeError(f"missing or invalid cumulative status counter: {name}") return values def status_counter_snapshot(rust): if rust is None: return None status, _, body = rust.admin("/api/status") if status != 200: raise RuntimeError(f"{rust.label} status capture failed: HTTP {status}") return counter_values(json.loads(body)) def drop_counter_report(before, after, requests): if requests <= 0: raise ValueError("drop-rate denominator must contain completed requests") if (before is None) != (after is None): raise RuntimeError("both status snapshots are required for a drop delta") if before is not None: validate_drop_counters(before) validate_drop_counters(after) deltas = {name: after[name] - before[name] if before is not None else None for name in DROP_COUNTER_FIELDS} if any(value is not None and value < 0 for value in deltas.values()): raise RuntimeError("cumulative drop counter decreased during measured interval") fields = {field: deltas[name] for name, field in DROP_COUNTER_FIELDS.items()} fields["history_drop_percent_of_completed_requests"] = ( 100 * deltas["logs_dropped"] / requests if before is not None else None) return {"applicable": before is not None, "counter_deltas": deltas, "row_fields": fields, "completed_requests_denominator": requests, "denominator_is_approximate": True, "interval_note": DROP_INTERVAL_NOTE} def analytics_counter_report(before, after, requests): if requests <= 0 or (before is None) != (after is None): raise ValueError("analytics delta requires paired snapshots and completed requests") delta = traffic_analytics_delta(before, after) if before is not None else { "queue_dropped_delta": None, "recorded_requests_delta": None, "interval_seconds": None} fields = {"analytics_queue_dropped_delta": delta["queue_dropped_delta"], "analytics_recorded_requests_delta": delta["recorded_requests_delta"], "analytics_queue_drop_percent_of_completed_requests": ( 100 * delta["queue_dropped_delta"] / requests if before is not None else None)} return {"applicable": before is not None, **delta, "row_fields": fields, "completed_requests_denominator": requests, "denominator_is_approximate": True, "interval_note": DROP_INTERVAL_NOTE, "recorded_requests_note": "Asynchronous analytics records; not HTTP success or completion counts."} def persist_rows(out, rows): write_json(out / "runs.json", rows) if not rows: return with (out / "runs.csv").open("w", newline="") as stream: writer = csv.DictWriter(stream, fieldnames=list(rows[0])) writer.writeheader() writer.writerows(rows) groups = {} for row in rows: groups.setdefault((row["mode"], row["size"], row["concurrency"], row["variant"]), []).append(row) summary = [] for (mode, size, concurrency, variant), group in sorted(groups.items()): rps = [row["rps"] for row in group] summary.append({"mode": mode, "size": size, "concurrency": concurrency, "variant": variant, "samples": len(group), "rps_median": statistics.median(rps), "rps_min": min(rps), "rps_max": max(rps), "rps_cv_percent": statistics.stdev(rps) / statistics.mean(rps) * 100 if len(rps) > 1 else None, "p50_us_median": statistics.median(row["p50_us"] for row in group), "p95_us_median": statistics.median(row["p95_us"] for row in group), "p99_us_median": statistics.median(row["p99_us"] for row in group), "server_cpu_us_per_request_median": statistics.median(row["server_cpu_us_per_request"] for row in group), "server_rss_peak_bytes": max(row["server_rss_peak_bytes"] for row in group), "server_fd_peak": max(row["server_fd_peak"] for row in group), "requests": sum(row["requests"] for row in group), "cgroup_throttled_samples": sum(row["cgroup_nr_throttled_delta"] > 0 for row in group), "cgroup_nr_throttled_delta": sum(row["cgroup_nr_throttled_delta"] for row in group), "cgroup_throttled_usec_delta": sum(row["cgroup_throttled_usec_delta"] for row in group), "validation_failures": sum(not (row["payload_before_passed"] and row["payload_after_passed"] and row["transfer_sanity_passed"]) for row in group), "errors": sum(row["error_total"] for row in group)}) write_json(out / "summary.json", summary) with (out / "summary.csv").open("w", newline="") as stream: writer = csv.DictWriter(stream, fieldnames=list(summary[0])) writer.writeheader() writer.writerows(summary) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--binary", action="append", required=True, metavar="LABEL=/ABS/PATH") parser.add_argument("--max-idle", action="append", default=[], metavar="LABEL=COUNT") parser.add_argument("--nginx", type=Path, required=True) parser.add_argument("--wrk", type=Path, required=True) parser.add_argument("--out", type=Path, required=True) parser.add_argument("--workers", type=int, default=4) parser.add_argument("--origin-workers", type=int, default=4) parser.add_argument("--client-threads", type=int, default=4) parser.add_argument("--server-cpus") parser.add_argument("--origin-cpus") parser.add_argument("--client-cpus") parser.add_argument("--duration", type=int, default=15) parser.add_argument("--warmup", type=int, default=3) parser.add_argument("--repeats", type=int, default=5) parser.add_argument("--concurrency", default="64") parser.add_argument("--sizes", default="small,large") parser.add_argument("--modes", default="proxy,static") parser.add_argument("--resource-interval", type=float, default=.5) parser.add_argument("--exploratory", action="store_true", help="explicitly allow <5 repetitions or <15s samples") parser.add_argument("--prepare-only", action="store_true", help="start processes and verify hashes; run no load generator") args = parser.parse_args() if (args.duration < 15 or args.repeats < 5) and not args.exploratory: parser.error("acceptance requires >=15-second runs and >=5 repetitions; use --exploratory for labelled short diagnostics") if args.out.exists() and any(args.out.iterdir()): parser.error("--out must be a new or empty directory so raw evidence is never overwritten") modes, sizes = args.modes.split(","), args.sizes.split(",") concurrencies = [int(value) for value in args.concurrency.split(",")] if any(mode not in ("proxy", "static") for mode in modes) or any(size not in SIZES for size in sizes): parser.error("supported modes: proxy,static; sizes: small,large,huge") if min(concurrencies) < 1 or max(concurrencies) > 4096: parser.error("concurrency must be 1..4096") if args.client_threads < 1: parser.error("--client-threads must be positive") if args.workers < 1 or args.origin_workers < 1: parser.error("--workers and --origin-workers must be positive") try: cpu_sets = validate_cpu_sets({"server": args.server_cpus, "origin": args.origin_cpus, "client": args.client_cpus}, set(os.sched_getaffinity(0))) except ValueError as error: parser.error(str(error)) if args.resource_interval <= 0: parser.error("--resource-interval must be positive") args.nginx, args.wrk, args.out = args.nginx.resolve(), args.wrk.resolve(), args.out.resolve() variants = dict(item.split("=", 1) for item in args.binary) if len(variants) != len(args.binary) or "nginx" in variants: parser.error("variant names must be unique and must not be nginx") max_idle = dict(item.split("=", 1) for item in args.max_idle) args.out.mkdir(parents=True, exist_ok=True) environment = {"kind": "same-host real HTTP/1.1 process benchmark", "simulation": False, "exploratory": args.exploratory, "prepare_only": args.prepare_only, "acceptance_sampling": args.duration >= 15 and args.repeats >= 5 and not args.prepare_only, "platform": platform.platform(), "python": sys.version, "started_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "cpu_affinity_available": sorted(os.sched_getaffinity(0)), "cpu_sets": {label: sorted(cpus) for label, cpus in cpu_sets.items()}, "resource_process_discovery": "PPid map from /proc/*/stat; root and configured nginx process counts must be readable", "expected_nginx_processes": {"frontend": args.workers + 1, "origin": args.origin_workers + 1}, "generator_plan": [{"connections": concurrency, "requested_threads": args.client_threads, "effective_threads": effective_threads(args.client_threads, concurrency)} for concurrency in concurrencies], "settings": {key: str(value) if isinstance(value, Path) else value for key, value in vars(args).items()}, "wrk_sha256": sha256(args.wrk), "nginx_sha256": sha256(args.nginx), "limitations": ["Generator, upstream and frontend share one host and kernel; not production capacity", "Closed-loop wrk latency includes its histogram correction; no open-loop arrival-rate SLA claim", "RSS sums processes and may double-count shared pages in nginx", "RSS and FD peaks are sampled, not absolute instantaneous peaks", "Exact HTTP 200 and payload hashes are checked before/after each measured run, not for every response during load", "wrk status errors count only status >399; aggregate bytes/request is a gross sanity check, not response integrity proof", "cgroup CPU-throttling deltas cover all processes in the shared cgroup, not only the measured server", DROP_INTERVAL_NOTE, "No TLS/H2/H3 speed inference; no CPU cycles/allocation profiler claim", "nginx has a smaller feature set; request logging/ACL/analytics work is not feature-matched"]} for path in ["/sys/fs/cgroup/cpu.max", "/sys/fs/cgroup/memory.max", "/proc/version"]: try: environment[path] = Path(path).read_text().strip() except OSError: environment[path] = None write_json(args.out / "environment.json", environment) expected_samples = len(modes) * len(sizes) * len(concurrencies) * args.repeats * (len(variants) + 1) write_json(args.out / "completion.json", {"completed": False, "samples": 0, "expected_samples": expected_samples, "simulation": False, "prepare_only": args.prepare_only}) rows, error = [], None try: with Fixture(args.out / "fixture", workers=args.workers, server_cpus=args.server_cpus, origin_cpus=args.origin_cpus, origin_workers=args.origin_workers) as fixture: nginx = fixture.nginx(args.nginx) servers = {"nginx": {"process": nginx["process"], "proxy": nginx["reverse_port"], "static": nginx["static_port"], "rust": None}} for label, binary in variants.items(): rust = fixture.rust(Path(binary), label, max_idle=int(max_idle[label]) if label in max_idle else None) rust.login() servers[label] = {"process": rust.process, "proxy": rust.reverse_port, "static": rust.static_port, "rust": rust} quiescence = {label: server["rust"].wait_quiescent() for label, server in servers.items() if server["rust"] is not None} write_json(args.out / "quiescence-preflight.json", quiescence) initial_affinity = {"origin": affinity_snapshot(nginx["origin"].pid, cpu_sets["origin"], args.origin_workers + 1)} initial_affinity.update({label: affinity_snapshot(server["process"].pid, cpu_sets["server"], args.workers + 1 if label == "nginx" else 1) for label, server in servers.items()}) write_json(args.out / "affinity-preflight.json", initial_affinity) # The fixture has only master + configured workers. Refuse a # master-only snapshot even if HTTP payload checks would succeed. prepare_monitor = Monitor({"server": nginx["process"].pid, "origin": nginx["origin"].pid}, args.resource_interval, {"server": args.workers + 1, "origin": args.origin_workers + 1}) prepare_monitor.sample() write_json(args.out / "resources-preflight.json", prepare_monitor.samples) telemetry_preflight = {label: {"status": status_counter_snapshot(server["rust"]), "traffic": traffic_analytics_snapshot(server["rust"]) if server["rust"] else None} for label, server in servers.items()} write_json(args.out / "telemetry-preflight.json", telemetry_preflight) for snapshot in telemetry_preflight.values(): if snapshot["status"] is not None: validate_drop_counters(snapshot["status"]) preflight = [] for label, server in servers.items(): for mode, size in itertools.product(modes, sizes): payload = (fixture.site / SIZES[size]).read_bytes() check = payload_check(server[mode], SIZES[size], payload) preflight.append({"variant": label, "mode": mode, "size": size, **check}) write_json(args.out / "preflight.json", preflight) if not check["passed"]: raise AssertionError(f"payload preflight failed: {label} {mode} {size}: {check}") write_json(args.out / "preflight.json", preflight) labels = [*variants, "nginx"] schedule = [] for mode, size, concurrency in itertools.product(modes, sizes, concurrencies): for repetition in range(args.repeats): offset = repetition % len(labels) order = labels[offset:] + labels[:offset] if repetition % 2: order.reverse() for variant in order: schedule.append({"sequence": len(schedule) + 1, "mode": mode, "size": size, "concurrency": concurrency, "repetition": repetition + 1, "variant": variant}) write_json(args.out / "schedule.json", schedule) if not args.prepare_only: for item in schedule: server = servers[item["variant"]] expected_processes = {"origin": args.origin_workers + 1, "server": args.workers + 1 if item["variant"] == "nginx" else 1} stem = f"{item['sequence']:04}-{item['mode']}-{item['size']}-c{item['concurrency']}-{item['variant']}" if args.warmup: warmup = wrk_run(args, server[item["mode"]], SIZES[item["size"]], args.warmup, item["concurrency"], args.out / "warmup" / stem) if warmup["error_total"]: raise RuntimeError(f"warmup errors in {stem}; raw evidence retained") payload = (fixture.site / SIZES[item["size"]]).read_bytes() before = payload_check(server[item["mode"]], SIZES[item["size"]], payload) write_json(args.out / "checks" / (stem + "-before.json"), before) if not before["passed"]: raise RuntimeError(f"exact payload check failed before {stem}: {before}") counters_before = status_counter_snapshot(server["rust"]) write_json(args.out / "status" / (stem + "-before.json"), counters_before) if counters_before is not None: validate_drop_counters(counters_before) analytics_before = traffic_analytics_snapshot(server["rust"]) if server["rust"] else None write_json(args.out / "status" / (stem + "-traffic-before.json"), analytics_before) cpu_before = cgroup_cpu_stat() try: result = wrk_run(args, server[item["mode"]], SIZES[item["size"]], args.duration, item["concurrency"], args.out / "raw" / stem, {"server": server["process"].pid, "origin": nginx["origin"].pid}, expected_processes) finally: cpu_after = cgroup_cpu_stat() cpu_delta = cgroup_cpu_delta(cpu_before, cpu_after) throttling = {"before": cpu_before, "after": cpu_after, "delta": cpu_delta} write_json(args.out / "checks" / (stem + "-cgroup-cpu.json"), throttling) after = payload_check(server[item["mode"]], SIZES[item["size"]], payload) write_json(args.out / "checks" / (stem + "-after.json"), after) write_json(args.out / "checks" / (stem + "-affinity-after.json"), { "server": affinity_snapshot(server["process"].pid, cpu_sets["server"], expected_processes.get("server")), "origin": affinity_snapshot(nginx["origin"].pid, cpu_sets["origin"], expected_processes["origin"])}) counters_after = status_counter_snapshot(server["rust"]) write_json(args.out / "status" / (stem + ".json"), counters_after) analytics_after = traffic_analytics_snapshot(server["rust"]) if server["rust"] else None write_json(args.out / "status" / (stem + "-traffic-after.json"), analytics_after) drops = drop_counter_report(counters_before, counters_after, result["requests"]) write_json(args.out / "status" / (stem + "-delta.json"), drops) analytics = analytics_counter_report(analytics_before, analytics_after, result["requests"]) write_json(args.out / "status" / (stem + "-traffic-delta.json"), analytics) sanity = transfer_sanity(result, len(payload), before["approximate_header_bytes"], item["concurrency"]) result.update({"payload_before": before, "payload_after": after, "transfer_sanity": sanity, "cgroup_cpu": throttling, "observability": drops, "traffic_analytics": analytics}) write_json((args.out / "raw" / stem).with_suffix(".json"), result) row = {**item, **{key: result[key] for key in ["requests", "bytes", "duration_us", "rps", "p50_us", "p95_us", "p99_us", "client_threads_requested", "client_threads", "requested_connections", "effective_connections", "error_total", "errors_connect", "errors_read", "errors_write", "errors_timeout", "errors_status", "generator_cpu_seconds"]}} row.update({"payload_before_passed": before["passed"], "payload_after_passed": after["passed"], "transfer_sanity_passed": sanity["passed"], "wire_bytes_per_request": sanity["wire_bytes_per_request"], "cgroup_nr_throttled_delta": cpu_delta["nr_throttled"], "cgroup_throttled_usec_delta": cpu_delta["throttled_usec"]}) row.update(drops["row_fields"]) row.update(analytics["row_fields"]) for group in ["server", "origin"]: row.update({f"{group}_{key}": value for key, value in result["resources"][group].items()}) rows.append(row) persist_rows(args.out, rows) print(json.dumps({"sequence": item["sequence"], "total": len(schedule), "variant": item["variant"], "scenario": f"{item['mode']}_{item['size']}_c{item['concurrency']}", "rps": round(result["rps"], 2), "p95_ms": result["p95_us"] / 1000, "errors": result["error_total"]}), flush=True) if result["error_total"]: raise RuntimeError(f"benchmark errors in {stem}; raw evidence retained") if not after["passed"] or not sanity["passed"]: raise RuntimeError(f"payload/byte-count validation failed in {stem}; raw evidence retained") except (Exception, KeyboardInterrupt) as exc: error = {"error": str(exc) or type(exc).__name__, "type": type(exc).__name__, "interrupted": isinstance(exc, KeyboardInterrupt), "traceback": traceback.format_exc()} write_json(args.out / "failure.json", error) print(json.dumps(error), flush=True) finally: persist_rows(args.out, rows) completed = error is None and (args.prepare_only or len(rows) == expected_samples) write_json(args.out / "completion.json", {"completed": completed, "samples": len(rows), "expected_samples": expected_samples, "simulation": False, "prepare_only": args.prepare_only, "finished_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())}) seal_results(args.out) return (130 if error.get("interrupted") else 1) if error else (0 if completed else 1) if __name__ == "__main__": sys.exit(main())