#!/usr/bin/env python3 """Build the NS1-host study report (charts + markdown + HTML + PDF) from a results dir. Must be able to run entirely on NS1.GEORGELAMBERT.ORG with python3, matplotlib, pandoc, and weasyprint. Parses nats bench logs; does not hard-code rates. """ from __future__ import annotations import json import re import subprocess import sys from pathlib import Path ROOT = Path(__file__).resolve().parents[1] import importlib.util _spec = importlib.util.spec_from_file_location( "bench_report", Path(__file__).resolve().parent / "bench-report.py" ) _br = importlib.util.module_from_spec(_spec) assert _spec.loader is not None _spec.loader.exec_module(_br) fmt_int = _br.fmt_int lat_mode = _br.lat_mode lat_sort = _br.lat_sort parse_bench = _br.parse_bench parse_lat = _br.parse_lat thru_sort = _br.thru_sort try: import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt from matplotlib.ticker import FuncFormatter except ImportError as e: raise SystemExit(f"matplotlib required on NS1: {e}") from e INDIGO = "#4f46e5" DEEP = "#312e81" TEAL = "#047857" AMBER = "#b45309" LILAC = "#7c74f0" INK = "#171a26" MUTED = "#5b6178" GRID = "#d9dce8" CORE_LABELS = { "core-1p1s-50k-128": "1p1s\n50k×128 B", "core-4p4s-100k-128": "4p4s\n100k×128 B", "core-8p8s-200k-128": "8p8s\n200k×128 B", "core-4p4s-50k-1k": "4p4s\n50k×1 KiB", } JS_LABELS = { "js-1p-20k-128-r3": "1p 20k×128 B", "js-4p-50k-128-r3": "4p 50k×128 B", "js-4p-20k-1k-r3": "4p 20k×1 KiB", "js-2p2s-20k-128-r3": "2p2s pull 20k×128 B", "js-mem-1p-20k-128-r3": "mem 1p 128 B", "js-mem-4p-50k-128-r3": "mem 4p 128 B", "js-mem-4p-20k-1k-r3": "mem 4p 1 KiB", } LAT_LABELS = { "lat-ping-1k-128": "Ping\n1k×128 B", "lat-1p-5k-128": "Flood 1p\n5k×128 B", "lat-4p-5k-1k": "Flood 4p\n5k×1 KiB", "lat-4p-10k-128": "Flood 4p\n10k×128 B", "lat-8p-20k-128": "Flood 8p\n20k×128 B", } def ms(s: str) -> float: return float(s.replace("ms", "").replace(",", "").strip()) def k_fmt(x: float, _pos: int | None = None) -> str: if x >= 1_000_000: return f"{x / 1_000_000:.2f}M" if x >= 1000: return f"{x / 1000:.0f}k" return f"{x:.0f}" def style() -> None: plt.rcParams.update( { "font.family": "sans-serif", "font.size": 10, "axes.titlesize": 12, "axes.titleweight": "semibold", "axes.edgecolor": GRID, "axes.labelcolor": INK, "text.color": INK, "xtick.color": MUTED, "ytick.color": MUTED, "figure.facecolor": "white", "axes.facecolor": "white", "axes.grid": True, "grid.color": GRID, "grid.linewidth": 0.8, "legend.frameon": False, "savefig.bbox": "tight", "savefig.dpi": 160, "savefig.facecolor": "white", } ) def save(fig: plt.Figure, path: Path) -> None: fig.savefig(path, dpi=160) plt.close(fig) def load_runs(folder: Path) -> tuple[list[dict[str, str]], list[dict[str, str]]]: thru: list[dict[str, str]] = [] lats: list[dict[str, str]] = [] for f in sorted(folder.glob("*.txt")): if f.name.startswith("host-"): continue text = f.read_text(encoding="utf-8", errors="replace") lat = parse_lat(text) if lat: lat["run"] = f.stem lats.append(lat) continue p = parse_bench(text) if p.get("pub_msgs") or p.get("agg_msgs"): p["run"] = f.stem thru.append(p) return sorted(thru, key=thru_sort), sorted(lats, key=lat_sort) def kv_file(path: Path) -> dict[str, str]: out: dict[str, str] = {} if not path.exists(): return out for line in path.read_text(encoding="utf-8", errors="replace").splitlines(): if "=" in line and not line.startswith("---"): k, _, v = line.partition("=") if k.strip() in out: continue out[k.strip()] = v.strip() return out def thru_table(thru: list[dict[str, str]]) -> str: lines = [ "| Run | Mode | Aggregate msgs/s | Pub msgs/s | Pub MB/s | Sub msgs/s | Sub MB/s |", "|-----|------|------------------|------------|----------|------------|----------|", ] for p in thru: lines.append( f"| `{p['run']}` | {p.get('mode', '')} | {fmt_int(p.get('agg_msgs'))} | " f"{fmt_int(p.get('pub_msgs'))} | {p.get('pub_mb') or '—'} | " f"{fmt_int(p.get('sub_msgs'))} | {p.get('sub_mb') or '—'} |" ) return "\n".join(lines) def delay_table(lats: list[dict[str, str]]) -> str: lines = [ "| Run | Kind | Count | Pubs | Size | min | avg | p50 | p90 | p99 | max |", "|-----|------|-------|------|------|-----|-----|-----|-----|-----|-----|", ] for p in lats: kind = lat_mode(p.get("run", ""), p.get("mode", "")) label = "ping (sequential RTT)" if kind == "ping" else "flood (burst queueing)" lines.append( f"| `{p['run']}` | {label} | {p.get('count', '')} | {p.get('pubs', '')} | " f"{p.get('size', '')} B | {p.get('min', '')} | {p.get('avg', '')} | " f"{p.get('p50', '')} | {p.get('p90', '')} | {p.get('p99', '')} | {p.get('max', '')} |" ) return "\n".join(lines) def varz_table(path: Path) -> str: if not path.exists(): return "_varz snapshot not captured._" rows = json.loads(path.read_text(encoding="utf-8")) lines = [ "| Node | VMID | connections | in_msgs | out_msgs | cpu | cores | mem (B) | jetstream |", "|------|------|-------------|---------|----------|-----|-------|---------|-----------|", ] for r in rows: if r.get("error"): lines.append(f"| {r.get('name')} | {r.get('vmid')} | error: {r['error']} | | | | | | |") continue lines.append( f"| {r.get('name')} | {r.get('vmid')} | {r.get('connections')} | " f"{r.get('in_msgs'):,} | {r.get('out_msgs'):,} | {r.get('cpu')} | " f"{r.get('cores')} | {r.get('mem'):,} | {r.get('jetstream')} |" ) return "\n".join(lines) def charts(thru: list[dict[str, str]], lats: list[dict[str, str]], dest: Path) -> None: dest.mkdir(parents=True, exist_ok=True) style() by = {p["run"]: p for p in thru} core_keys = [k for k in CORE_LABELS if k in by] if core_keys: fig, ax = plt.subplots(figsize=(9.2, 4.4)) x = list(range(len(core_keys))) w = 0.25 agg = [int(by[k].get("agg_msgs") or 0) for k in core_keys] pub = [int(by[k].get("pub_msgs") or 0) for k in core_keys] sub = [int(by[k].get("sub_msgs") or 0) for k in core_keys] ax.bar([i - w for i in x], agg, w, label="Aggregate", color=DEEP) ax.bar(x, pub, w, label="Publish", color=INDIGO) ax.bar([i + w for i in x], sub, w, label="Subscribe", color=TEAL) ax.set_xticks(x, [CORE_LABELS[k] for k in core_keys]) ax.set_ylabel("messages / second") ax.set_title("Core NATS throughput (fire-and-forget) — NS1 host run") ax.yaxis.set_major_formatter(FuncFormatter(k_fmt)) ax.legend(loc="upper left") ax.set_axisbelow(True) save(fig, dest / "core-throughput.png") js_keys = [k for k in JS_LABELS if k in by] if js_keys: fig, ax = plt.subplots(figsize=(9.2, 4.4)) pubs = [int(by[k].get("pub_msgs") or 0) for k in js_keys] colors = [INDIGO, INDIGO, AMBER, LILAC][: len(js_keys)] ax.bar([JS_LABELS[k] for k in js_keys], pubs, color=colors) ax.set_ylabel("durable publish messages / second") ax.set_title("JetStream file store, replicas=3 — NS1 host run") ax.yaxis.set_major_formatter(FuncFormatter(k_fmt)) ax.set_axisbelow(True) for i, v in enumerate(pubs): ax.text(i, v * 1.02, f"{v:,}", ha="center", va="bottom", fontsize=9, color=MUTED) save(fig, dest / "js-throughput.png") pair = [("core-1p1s-50k-128", "js-1p-20k-128-r3"), ("core-4p4s-100k-128", "js-4p-50k-128-r3"), ("core-4p4s-50k-1k", "js-4p-20k-1k-r3")] if all(c in by and j in by for c, j in pair): fig, ax = plt.subplots(figsize=(9.2, 4.4)) labels = ["1 publisher\n128 B", "4 publishers\n128 B", "4 publishers\n1 KiB"] core_pub = [int(by[c]["pub_msgs"]) for c, _ in pair] js_pub = [int(by[j]["pub_msgs"]) for _, j in pair] x = list(range(3)) w = 0.35 ax.bar([i - w / 2 for i in x], core_pub, w, label="Core NATS (no disk)", color=INDIGO) ax.bar([i + w / 2 for i in x], js_pub, w, label="JetStream r=3 file", color=AMBER) ax.set_xticks(x, labels) ax.set_yscale("log") ax.set_ylabel("publish messages / second (log)") ax.set_title("Core vs JetStream — NS1 host run") ax.legend(loc="upper right") ax.set_axisbelow(True) save(fig, dest / "core-vs-js.png") if "core-4p4s-100k-128" in by and "core-4p4s-50k-1k" in by: fig, axes = plt.subplots(1, 2, figsize=(9.2, 4.2)) labels = ["128 B\n4p4s", "1 KiB\n4p4s"] msgs = [int(by["core-4p4s-100k-128"].get("agg_msgs") or 0), int(by["core-4p4s-50k-1k"].get("agg_msgs") or 0)] mb = [float(by["core-4p4s-100k-128"].get("agg_mb") or 0), float(by["core-4p4s-50k-1k"].get("agg_mb") or 0)] axes[0].bar(labels, msgs, color=[INDIGO, AMBER]) axes[0].set_title("Aggregate messages / second") axes[0].yaxis.set_major_formatter(FuncFormatter(k_fmt)) axes[1].bar(labels, mb, color=[INDIGO, AMBER]) axes[1].set_title("Aggregate MB / second") fig.suptitle("Core NATS payload effect — NS1 host run", fontsize=12, fontweight="semibold") fig.tight_layout() save(fig, dest / "payload-size.png") if lats: fig, ax = plt.subplots(figsize=(9.2, 4.6)) ordered = [p for p in lats] labels = [LAT_LABELS.get(p["run"], p["run"]) for p in ordered] x = list(range(len(ordered))) w = 0.25 p50 = [ms(p["p50"]) for p in ordered] p90 = [ms(p["p90"]) for p in ordered] p99 = [ms(p["p99"]) for p in ordered] ax.bar([i - w for i in x], p50, w, label="p50", color=TEAL) ax.bar(x, p90, w, label="p90", color=INDIGO) ax.bar([i + w for i in x], p99, w, label="p99", color=AMBER) ax.set_xticks(x, labels) ax.set_yscale("log") ax.set_ylabel("milliseconds (log)") ax.set_title("Round-trip delay — NS1 host run") ax.axhline(1.0, color=GRID, linestyle="--", linewidth=1) ax.legend(loc="upper left") ax.set_axisbelow(True) save(fig, dest / "delay-percentiles.png") def figure(name: str, caption: str) -> str: return f"![{caption}](charts/{name})\n\n*{caption}*" def ratio(new: str | None, old: str | None) -> str: if not new or not old: return "—" a, b = float(new), float(old) if b == 0: return "—" return f"{a / b:.2f}×" def delay_ms_val(p: dict[str, str] | None, key: str) -> str | None: if not p or not p.get(key): return None return str(ms(p[key])) def delta_table( thru: list[dict[str, str]], lats: list[dict[str, str]], base_thru: list[dict[str, str]], base_lats: list[dict[str, str]], base_stamp: str, ) -> str: bt = {p["run"]: p for p in base_thru} nt = {p["run"]: p for p in thru} bl = {p["run"]: p for p in base_lats} nl = {p["run"]: p for p in lats} keys = [ ("core-1p1s-50k-128", "pub", "Core 1p1s 128 B pub msgs/s"), ("core-8p8s-200k-128", "agg", "Core 8p8s 128 B aggregate msgs/s"), ("js-1p-20k-128-r3", "pub", "JS file r=3 1p 128 B pub msgs/s"), ("js-4p-50k-128-r3", "pub", "JS file r=3 4p 128 B pub msgs/s"), ("js-4p-20k-1k-r3", "pub", "JS file r=3 4p 1 KiB pub msgs/s"), ("js-mem-1p-20k-128-r3", "pub", "JS memory r=3 1p 128 B pub msgs/s"), ("js-mem-4p-50k-128-r3", "pub", "JS memory r=3 4p 128 B pub msgs/s"), ] lines = [ f"| Metric | Baseline `{base_stamp}` | This run | Ratio |", "|--------|-------------------------|----------|-------|", ] for run, kind, label in keys: old, new = bt.get(run), nt.get(run) ok = "pub_msgs" if kind == "pub" else "agg_msgs" ov = old.get(ok) if old else None nv = new.get(ok) if new else None lines.append(f"| {label} | {fmt_int(ov)} | {fmt_int(nv)} | {ratio(nv, ov)} |") old_p, new_p = bl.get("lat-ping-1k-128"), nl.get("lat-ping-1k-128") if old_p or new_p: ov = delay_ms_val(old_p, "p99") nv = delay_ms_val(new_p, "p99") # smaller delay is better — invert ratio label r = "—" if ov and nv and float(nv) != 0: r = f"{float(ov) / float(nv):.2f}× faster" if float(nv) < float(ov) else f"{float(nv) / float(ov):.2f}× slower" lines.append( f"| Ping p99 (ms) | {old_p.get('p99') if old_p else '—'} | {new_p.get('p99') if new_p else '—'} | {r} |" ) return "\n".join(lines) def chart_delta( thru: list[dict[str, str]], base_thru: list[dict[str, str]], dest: Path, ) -> None: bt = {p["run"]: p for p in base_thru} nt = {p["run"]: p for p in thru} labels = ["Core 1p\n128 B pub", "JS file 1p\n128 B", "JS file 4p\n128 B", "JS mem 1p\n128 B"] keys = ["core-1p1s-50k-128", "js-1p-20k-128-r3", "js-4p-50k-128-r3", "js-mem-1p-20k-128-r3"] old = [int(bt[k]["pub_msgs"]) if k in bt and bt[k].get("pub_msgs") else 0 for k in keys] new = [int(nt[k]["pub_msgs"]) if k in nt and nt[k].get("pub_msgs") else 0 for k in keys] if not any(new): return fig, ax = plt.subplots(figsize=(9.2, 4.4)) x = list(range(len(labels))) w = 0.35 ax.bar([i - w / 2 for i in x], old, w, label="Baseline 1c/1G/ZFS", color=MUTED) ax.bar([i + w / 2 for i in x], new, w, label="8c/16G/tmpfs (+ mem rows)", color=INDIGO) ax.set_xticks(x, labels) ax.set_yscale("log") ax.set_ylabel("publish messages / second (log)") ax.set_title("Measured delta vs 20260912T051237Z") ax.legend(loc="upper right") ax.set_axisbelow(True) save(fig, dest / "delta-vs-baseline.png") def write_markdown( folder: Path, thru: list[dict[str, str]], lats: list[dict[str, str]], compare: Path | None = None, ) -> str: before = kv_file(folder / "host-before.txt") after = kv_file(folder / "host-after.txt") stamp = folder.name method = (Path(__file__).resolve().parent / "ns1-study-methodology.md").read_text(encoding="utf-8") delta_md = "" base_thru: list[dict[str, str]] = [] base_lats: list[dict[str, str]] = [] if compare and compare.is_dir(): base_thru, base_lats = load_runs(compare) delta_md = ( f"## Measured delta vs `{compare.name}`\n\n" "Baseline: 1 core / 1 GiB / JetStream on ZFS. This run: 8 cores / 16 GiB / " "JetStream **tmpfs** (file r=3) plus extra **memory** store rows. veth/10G unchanged.\n\n" + delta_table(thru, lats, base_thru, base_lats, compare.name) + "\n" ) if (folder / "charts" / "delta-vs-baseline.png").exists(): delta_md += "\n" + figure("delta-vs-baseline.png", "Baseline vs maximized publish rates (log)") delta_md += "\n" figs = [] charts_dir = folder / "charts" if (charts_dir / "core-throughput.png").exists(): figs.append("### Core NATS\n\n" + figure("core-throughput.png", "Core NATS throughput at four loads (NS1 host run)")) if (charts_dir / "payload-size.png").exists(): figs.append("### Payload size (core)\n\n" + figure("payload-size.png", "Core NATS 128 B vs 1 KiB (NS1 host run)")) if (charts_dir / "js-throughput.png").exists(): figs.append("### JetStream r=3 file\n\n" + figure("js-throughput.png", "JetStream durable publish rate (NS1 host run)")) if (charts_dir / "core-vs-js.png").exists(): figs.append("### Core vs JetStream\n\n" + figure("core-vs-js.png", "Core vs JetStream publish rate, log scale (NS1 host run)")) if (charts_dir / "delay-percentiles.png").exists(): figs.append("### Delay\n\n" + figure("delay-percentiles.png", "Ping vs flood delay percentiles, log scale (NS1 host run)")) ping = next((p for p in lats if "ping" in p.get("run", "")), None) js1 = next((p for p in thru if p["run"] == "js-1p-20k-128-r3"), None) core1 = next((p for p in thru if p["run"] == "core-1p1s-50k-128"), None) md = f"""**Progress report (maximized NS1 study)** · run `{stamp}` (UTC) > **Execution provenance.** Every process for this study ran on **NS1.GEORGELAMBERT.ORG** (`70.88.205.138`): `maximize-ns1-study.sh` (cores/RAM/`max_mem`/tmpfs), then `study-on-ns1.sh`, `nats bench`, `latency.mjs` (LXC 510), matplotlib, pandoc, weasyprint. Traffic stayed on `vmbr1`. veth/10G was **not** changed. After the ladder, JetStream was put back on ZFS and product streams were re-created; **8 cores / 16 GiB / max_mem 8G stay**. {delta_md} --- ## 1. Executive summary | Item | This NS1-host run | |------|-------------------| | Control plane | NS1.GEORGELAMBERT.ORG (`70.88.205.138`), user `{before.get("whoami", "marchon")}` | | Bench client | LXC {before.get("client_vmid", "510")} `verae-px-worker` | | Brokers | LXC 511/512/513 `nats-a/b/c` on `10.10.10.21–23` | | Client URL | `{before.get("nats_url", "")}` | | Host load before | `{before.get("loadavg", "n/a")}` | | Host load after | `{after.get("loadavg", "n/a")}` | | Core 1p1s 128 B pub | {fmt_int(core1.get("pub_msgs") if core1 else None)} msgs/s | | JetStream 1p 128 B r=3 | {fmt_int(js1.get("pub_msgs") if js1 else None)} durable pubs/s | | Ping p50 / p99 | {ping.get("p50") if ping else "—"} / {ping.get("p99") if ping else "—"} | Product traffic is the JetStream row. Ping is one-message delay. Flood is mailbox catch-up after a burst. --- ## 2. Where it ran (and where it did not) ```text Operator laptop ──ssh──► NS1.GEORGELAMBERT.ORG 70.88.205.138 study-on-ns1.sh python3 build-ns1-study-report.py sudo pct exec 510 ──► nats bench / latency.mjs │ ▼ vmbr1 10.10.10.21-23 :4222 ``` - **Did run on 138:** bash, python3, matplotlib, pandoc, weasyprint, `pct`, nats-server (in LXC), nats CLI and Node (in LXC 510). - **Did not run on the laptop:** no local `nats bench`, no local charting, no local WeasyPrint for this file. --- ## 3. Results (this run) ### Host and brokers **Before** {varz_table(folder / "varz-before.json")} **After** {varz_table(folder / "varz-after.json")} nproc={before.get("nproc", "?")} · uname=`{before.get("uname", "")}` ### Throughput {thru_table(thru)} ### Round-trip delay {delay_table(lats)} {chr(10).join(figs)} --- {method} --- ## 6. Reproducing this study On **NS1 only**: ```bash cd ~/verae-src/verae-nats-cluster bash scripts/study-on-ns1.sh ``` The script exits if `hostname` is not NS1. Outputs land in `results//` including `nats-cluster-bench-ns1.{{md,html,pdf}}` and `charts/`. Copy those into `zapier-decisions/reports/` for the progress repo and catalog. Raw logs for this run: `results/{stamp}/`. """ return md def render(md_path: Path, html_path: Path, pdf_path: Path) -> None: css = Path(__file__).resolve().parent / "docs-print.css" header = html_path.with_suffix(".hdr.html") banner = html_path.with_suffix(".ban.html") css_text = css.read_text(encoding="utf-8") if css.exists() else "" header.write_text(f"\n", encoding="utf-8") banner.write_text( '
' '' '
Verae Time × Zapier · progress report · maximized NS1 study
' "

NATS cluster message speed — maximized (RAM disk + 8 cores)

" '
packages/zapier-decisions/reports/nats-cluster-bench-ns1.md
' "
\n", encoding="utf-8", ) r = subprocess.run( [ "pandoc", str(md_path), "-o", str(html_path), "--standalone", f"--resource-path={md_path.parent}", "--highlight-style=breezedark", "--metadata=title=NATS cluster message speed — NS1 host study", f"--include-in-header={header}", f"--include-before-body={banner}", ], capture_output=True, text=True, ) header.unlink(missing_ok=True) banner.unlink(missing_ok=True) if r.returncode != 0: raise SystemExit(f"pandoc failed: {r.stderr[-800:]}") w = subprocess.run(["weasyprint", str(html_path), str(pdf_path)], capture_output=True, text=True) if w.returncode != 0: raise SystemExit(f"weasyprint failed: {w.stderr[-800:]}") def main() -> int: folder = Path(sys.argv[1] if len(sys.argv) > 1 else ".") compare = Path(sys.argv[2]) if len(sys.argv) > 2 and sys.argv[2] else None thru, lats = load_runs(folder) charts(thru, lats, folder / "charts") if compare and compare.is_dir(): base_thru, _base_lats = load_runs(compare) chart_delta(thru, base_thru, folder / "charts") md = write_markdown(folder, thru, lats, compare if compare and compare.is_dir() else None) md_path = folder / "nats-cluster-bench-ns1.md" md_path.write_text(md, encoding="utf-8") html_path = folder / "nats-cluster-bench-ns1.html" pdf_path = folder / "nats-cluster-bench-ns1.pdf" render(md_path, html_path, pdf_path) print(f"wrote {md_path}") print(f"wrote {html_path}") print(f"wrote {pdf_path}") return 0 if __name__ == "__main__": raise SystemExit(main())