master-zapier-plan-draft/packages/verae-nats-cluster/scripts/build-ns1-study-report.py

#!/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",
}
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 write_markdown(folder: Path, thru: list[dict[str, str]], lats: list[dict[str, str]]) -> 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")
    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 (second study)** · run `{stamp}` (UTC)

> **Execution provenance.** Every process for this study ran on **NS1.GEORGELAMBERT.ORG** (`70.88.205.138`): the orchestrator (`study-on-ns1.sh`), `nats bench`, `latency.mjs` (inside LXC 510 on this hypervisor), charting (`matplotlib`), and HTML/PDF (`pandoc` + `weasyprint`). The operator laptop did **not** publish, subscribe, draw charts, or render the PDF. Traffic stayed on `vmbr1` from LXC **510** to `nats-a/b/c` (**511–513**).

This is a full methodology write-up plus the numbers from that on-host run. The earlier report (`nats-cluster-bench`, run `20260912T045131Z`) used the same cluster but was **orchestrated and rendered off-box**. Use this document when you need “it was all run on 138.”

---

## 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/<utc>/` 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"<style>{css_text}</style>\n", encoding="utf-8")
    banner.write_text(
        '<div class="doc-banner">'
        '<nav class="site"><a href="/">zapier.georgelambert.org</a>'
        ' · <a href="/index-md.html">Markdown indexes</a></nav>'
        '<div class="kicker">Verae Time × Zapier · progress report · run on NS1.GEORGELAMBERT.ORG</div>'
        "<h1>NATS cluster message speed — NS1 host study</h1>"
        '<div class="source-path">packages/zapier-decisions/reports/nats-cluster-bench-ns1.md</div>'
        "</div>\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 ".")
    thru, lats = load_runs(folder)
    charts(thru, lats, folder / "charts")
    md = write_markdown(folder, thru, lats)
    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())