#!/usr/bin/env python3 """Run wrk on a separate AWS client; preserve samples, errors and resource counters. Requires Linux, Python 3, ethtool, wrk 4.2.0 and (only for --bucket) boto3. Targets/configuration are supplied as JSON; no credentials or infrastructure creation live here. """ import argparse import datetime import gzip import hashlib import http.client import json import os import pathlib import platform import re import resource import socket import ssl import statistics import subprocess import time def utc(): return datetime.datetime.now(datetime.timezone.utc).isoformat() def counters(): cpu = [int(x) for x in pathlib.Path("/proc/stat").read_text().splitlines()[0].split()[1:]] net = {} for line in pathlib.Path("/proc/net/dev").read_text().splitlines()[2:]: name, values = line.split(":") if name.strip() != "lo": v = [int(x) for x in values.split()] net[name.strip()] = { "rx_bytes": v[0], "tx_bytes": v[8], "rx_drop": v[3], "tx_drop": v[11], } ena = {} for name in net: p = subprocess.run(["ethtool", "-S", name], capture_output=True, text=True) ena[name] = {k: int(v) for k, v in re.findall(r"\s+(\w*allowance\w*): (\d+)", p.stdout)} return {"cpu": cpu, "network": net, "ena": ena, "load": list(os.getloadavg())} def validate(server, case): conn = http.client.HTTPConnection(server["ip"], 80, timeout=15) if case["scheme"] == "https": # Connect privately but validate the real public identity and full chain. conn.sock = ssl.create_default_context().wrap_socket( socket.create_connection((server["ip"], 443), timeout=15), server_hostname=server["host"], ) try: conn.request( "GET", case["path"], headers={"Host": server["host"], "Accept-Encoding": case["encoding"]}, ) response = conn.getresponse() body = response.read() encoding = response.getheader("Content-Encoding", "identity") decoded = gzip.decompress(body) if encoding == "gzip" else body digest = hashlib.sha256(decoded).hexdigest() if response.status != 200 or digest != case["sha256"] or encoding != case["encoding"]: raise RuntimeError( f"Invalid response from {server['host']}: {response.status}, {encoding}, {digest}" ) return { "status": response.status, "encoding": encoding, "wire_body_bytes": len(body), "decoded_bytes": len(decoded), "sha256": digest, } finally: conn.close() LUA = r"""done = function(summary, latency, requests) io.write(string.format('\nRESULT {"requests":%d,"duration_us":%d,"bytes":%d,"latency_mean_us":%.3f,"latency_p50_us":%d,"latency_p95_us":%d,"latency_p99_us":%d,"errors":{"connect":%d,"read":%d,"write":%d,"status":%d,"timeout":%d}}\n', summary.requests, summary.duration, summary.bytes, latency.mean, latency:percentile(50), latency:percentile(95), latency:percentile(99), summary.errors.connect, summary.errors.read, summary.errors.write, summary.errors.status, summary.errors.timeout)) end """ def main(): p = argparse.ArgumentParser(description=__doc__) p.add_argument("config", type=pathlib.Path) p.add_argument("--output", type=pathlib.Path, default=pathlib.Path("aws-results.json")) p.add_argument("--rounds", type=int, default=5) p.add_argument("--seconds", type=int, default=10) p.add_argument("--warmup", type=int, default=2) p.add_argument("--threads", type=int, default=4) p.add_argument("--connections", type=int, default=64) p.add_argument("--bucket") p.add_argument("--key", default="benchmark-results/aws-results.json") p.add_argument( "--continue-on-errors", action="store_true", help="Resource studies only: retain failed samples and continue; never a clean throughput run", ) a = p.parse_args() if ( min(a.rounds, a.seconds, a.warmup, a.threads, a.connections) < 1 or a.connections < a.threads ): p.error("All sizes/durations must be positive, with at least one connection per thread") config = json.loads(a.config.read_text()) a.output.parent.mkdir(parents=True, exist_ok=True) lua = a.output.with_suffix(".lua") lua.write_text(LUA) result = { "started": utc(), "complete": False, "configuration": config, "client": {"uname": platform.uname()._asdict(), "cpus": os.cpu_count()}, "method": { "tool": "wrk 4.2.0", "protocol": "HTTP/1.1, no pipelining", "rounds": a.rounds, "seconds": a.seconds, "warmup_seconds": a.warmup, "threads": a.threads, "connections": a.connections, "order": "Rotate workloads by round and servers by round + workload; sequential runs", "continue_on_errors": a.continue_on_errors, }, "samples": [], } s3 = None if a.bucket: import boto3 s3 = boto3.client("s3", region_name="us-east-2") def save(): raw = json.dumps(result, indent=2) a.output.write_text(raw) if s3: s3.put_object( Bucket=a.bucket, Key=a.key, Body=raw.encode(), ContentType="application/json" ) save() try: for round_index in range(a.rounds): cases = config["cases"] for case_index in range(len(cases)): case = cases[(case_index + round_index) % len(cases)] for server_index in range(len(config["servers"])): server = config["servers"][ (server_index + round_index + case_index) % len(config["servers"]) ] before_validation = validate(server, case) cmd = [ "wrk", f"-t{a.threads}", f"-c{a.connections}", "--timeout", "5s", "-H", "Host: " + server["host"], "-H", "Accept-Encoding: " + case["encoding"], "-s", str(lua), case["scheme"] + "://" + server["ip"] + case["path"], ] warm = subprocess.run( [*cmd, f"-d{a.warmup}s"], capture_output=True, text=True, timeout=a.warmup + 20, ) if warm.returncode: raise RuntimeError(warm.stderr + warm.stdout) started = utc() before = counters() usage_before = resource.getrusage(resource.RUSAGE_CHILDREN) t0 = time.monotonic() run = subprocess.run( [*cmd, f"-d{a.seconds}s", "--latency"], capture_output=True, text=True, timeout=a.seconds + 20, ) wall = time.monotonic() - t0 usage_after = resource.getrusage(resource.RUSAGE_CHILDREN) after = counters() match = re.search(r"^RESULT (.+)$", run.stdout, re.M) if run.returncode or not match: raise RuntimeError(run.stderr + run.stdout) sample = json.loads(match[1]) sample.update( { "round": round_index + 1, "server": server["name"], "case": case["id"], "started": started, "finished": utc(), "wall_seconds": wall, "requests_per_second": sample["requests"] / (sample["duration_us"] / 1e6), "client_cpu_cores": ( (usage_after.ru_utime + usage_after.ru_stime) - (usage_before.ru_utime + usage_before.ru_stime) ) / wall, "client_before": before, "client_after": after, "validation_before": before_validation, "validation_after": validate(server, case), "command": [*cmd, f"-d{a.seconds}s", "--latency"], "stdout": run.stdout, "stderr": run.stderr, } ) result["samples"].append(sample) save() print( f"{len(result['samples'])}/{a.rounds * len(cases) * len(config['servers'])} round={round_index + 1} {case['id']} {server['name']}: {sample['requests_per_second']:.0f} req/s; CPU={sample['client_cpu_cores']:.2f} cores; errors={sum(sample['errors'].values())}", flush=True, ) if any(sample["errors"].values()) and not a.continue_on_errors: raise RuntimeError( "Request errors invalidate the run; inspect the preserved sample" ) result["summary"] = {} for case in config["cases"]: summary = result["summary"][case["id"]] = {} for server in config["servers"]: rows = [ x for x in result["samples"] if x["case"] == case["id"] and x["server"] == server["name"] ] values = [x["requests_per_second"] for x in rows] summary[server["name"]] = { "median_rps": statistics.median(values), "min_rps": min(values), "max_rps": max(values), "median_p99_us": statistics.median(x["latency_p99_us"] for x in rows), } result["complete"] = True except Exception as exc: result["failure"] = str(exc) raise finally: result["finished"] = utc() save() if __name__ == "__main__": main()