Snapshot of verae-nats-cluster from zapier monorepo (NATS cluster bench)
This commit is contained in:
commit
8afd8165dd
41 changed files with 1054 additions and 0 deletions
176
scripts/bench-report.py
Executable file
176
scripts/bench-report.py
Executable file
|
|
@ -0,0 +1,176 @@
|
|||
#!/usr/bin/env python3
|
||||
"""Turn nats bench text logs + latency JSON into markdown."""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def fmt_int(s: str | None) -> str:
|
||||
if not s:
|
||||
return "—"
|
||||
return f"{int(s):,}"
|
||||
|
||||
|
||||
def parse_bench(text: str) -> dict[str, str]:
|
||||
out: dict[str, str] = {"kind": "throughput"}
|
||||
m = re.search(r"(?m)^\s*Pub stats:\s*([0-9,]+)\s*msgs/sec\s*~\s*([0-9.]+)\s*MB/sec", text)
|
||||
if m:
|
||||
out["pub_msgs"] = m.group(1).replace(",", "")
|
||||
out["pub_mb"] = m.group(2)
|
||||
m = re.search(r"(?m)^\s*Sub stats:\s*([0-9,]+)\s*msgs/sec\s*~\s*([0-9.]+)\s*MB/sec", text)
|
||||
if m:
|
||||
out["sub_msgs"] = m.group(1).replace(",", "")
|
||||
out["sub_mb"] = m.group(2)
|
||||
m = re.search(r"NATS Pub/Sub stats:\s*([0-9,]+)\s*msgs/sec\s*~\s*([0-9.]+)\s*MB/sec", text)
|
||||
if m:
|
||||
out["agg_msgs"] = m.group(1).replace(",", "")
|
||||
out["agg_mb"] = m.group(2)
|
||||
if "JetStream" in text or "--js" in text or "js-" in text:
|
||||
out["mode"] = "jetstream r=3 file"
|
||||
else:
|
||||
out["mode"] = "core pub/sub"
|
||||
# nats 0.1.6 prints min/avg/max as msgs/sec across publishers, not µs delay
|
||||
m = re.search(
|
||||
r"min\s+([0-9,]+)\s*\|\s*avg\s+([0-9,]+)\s*\|\s*max\s+([0-9,]+)\s*\|\s*stddev\s+([0-9,]+)\s*msgs",
|
||||
text,
|
||||
)
|
||||
if m:
|
||||
out["pub_spread"] = f"{m.group(1)}–{m.group(3)} (avg {m.group(2)})"
|
||||
return out
|
||||
|
||||
|
||||
def parse_lat(text: str) -> dict[str, str] | None:
|
||||
for line in text.splitlines():
|
||||
line = line.strip()
|
||||
if line.startswith("{") and "p99_us" in line:
|
||||
d = json.loads(line)
|
||||
mode = d.get("mode") or ""
|
||||
return {
|
||||
"kind": "latency",
|
||||
"count": str(d.get("count", "")),
|
||||
"pubs": str(d.get("pubs", "")),
|
||||
"size": str(d.get("size", "")),
|
||||
"mode": str(mode),
|
||||
"min": d.get("min", ""),
|
||||
"avg": d.get("avg", ""),
|
||||
"p50": d.get("p50", ""),
|
||||
"p90": d.get("p90", ""),
|
||||
"p99": d.get("p99", ""),
|
||||
"max": d.get("max", ""),
|
||||
}
|
||||
return None
|
||||
|
||||
|
||||
def lat_mode(run: str, recorded: str) -> str:
|
||||
if recorded in ("ping", "flood"):
|
||||
return recorded
|
||||
if "ping" in run:
|
||||
return "ping"
|
||||
return "flood"
|
||||
|
||||
|
||||
def thru_sort(p: dict[str, str]) -> tuple:
|
||||
return (0 if p.get("mode", "").startswith("core") else 1, p.get("run", ""))
|
||||
|
||||
|
||||
def lat_sort(p: dict[str, str]) -> tuple:
|
||||
mode = lat_mode(p.get("run", ""), p.get("mode", ""))
|
||||
return (0 if mode == "ping" else 1, int(p.get("count") or 0), p.get("run", ""))
|
||||
|
||||
|
||||
def main() -> int:
|
||||
folder = Path(sys.argv[1] if len(sys.argv) > 1 else ".")
|
||||
thru: list[dict[str, str]] = []
|
||||
lats: list[dict[str, str]] = []
|
||||
for f in sorted(folder.glob("*.txt")):
|
||||
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)
|
||||
|
||||
stamp = folder.name if re.fullmatch(r"\d{8}T\d{6}Z", folder.name) else ""
|
||||
|
||||
print("# NATS cluster message speed")
|
||||
print()
|
||||
if stamp:
|
||||
print(f"Run **`{stamp}`** (UTC). ", end="")
|
||||
print(
|
||||
"Client: LXC **510** `verae-px-worker` (`10.10.10.20`), not a nats-* server. "
|
||||
"Servers: `nats-a/b/c` on `10.10.10.21–23` (`vmbr1` only)."
|
||||
)
|
||||
print()
|
||||
print("Client URL:")
|
||||
print()
|
||||
print("```text")
|
||||
print("nats://10.10.10.21:4222,nats://10.10.10.22:4222,nats://10.10.10.23:4222")
|
||||
print("```")
|
||||
print()
|
||||
print("## Method")
|
||||
print()
|
||||
print("- **Core NATS** is fire-and-forget pub/sub (`nats bench`). No disk, no replica ack.")
|
||||
print(
|
||||
"- **JetStream** uses **file** storage and **replicas=3** (same as product streams). "
|
||||
"The unique stream `benchstream` is deleted between JS loads."
|
||||
)
|
||||
print("- Throughput is **msgs/sec** from nats CLI **0.1.6** (`--no-progress --csv`). Its min/avg/max are publisher **rate spread**, not delay.")
|
||||
print(
|
||||
"- **Ping** delay: one publisher, sequential publish-then-wait. This is one-message round-trip through the cluster."
|
||||
)
|
||||
print(
|
||||
"- **Flood** delay: N publishers dump the whole batch, then the subscriber drains. "
|
||||
"This is **queueing under burst**, not wire RTT."
|
||||
)
|
||||
print("- Probe: `scripts/latency.mjs` (two connections, header timestamp).")
|
||||
print()
|
||||
print("## Throughput")
|
||||
print()
|
||||
print("| Run | Mode | Aggregate msgs/s | Pub msgs/s | Pub MB/s | Sub msgs/s | Sub MB/s |")
|
||||
print("|-----|------|------------------|------------|----------|------------|----------|")
|
||||
for p in sorted(thru, key=thru_sort):
|
||||
print(
|
||||
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 '—'} |"
|
||||
)
|
||||
print()
|
||||
print("## Round-trip delay")
|
||||
print()
|
||||
print("| Run | Kind | Count | Pubs | Size | min | avg | p50 | p90 | p99 | max |")
|
||||
print("|-----|------|-------|------|------|-----|-----|-----|-----|-----|-----|")
|
||||
for p in sorted(lats, key=lat_sort):
|
||||
kind = lat_mode(p.get("run", ""), p.get("mode", ""))
|
||||
label = "ping (sequential RTT)" if kind == "ping" else "flood (burst queueing)"
|
||||
print(
|
||||
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', '')} |"
|
||||
)
|
||||
print()
|
||||
print("## What the numbers mean")
|
||||
print()
|
||||
print(
|
||||
"Product job/event/archive traffic is **JetStream r=3 file**. On this three-LXC stand that is about "
|
||||
"**16k durable 128 B pubs/s** (about **13k** at 1 KiB). Core NATS is an upper bound for "
|
||||
"non-durable fan-out: about **0.7–2.0M msgs/s** aggregate at 128 B, or **~630k msgs/s (~616 MB/s)** at 1 KiB with 4 publishers."
|
||||
)
|
||||
print()
|
||||
print(
|
||||
"A quiet request-reply is **~0.3 ms** average, **p99 < 1 ms**. Flood rows in the **150–500 ms** band "
|
||||
"are the subscriber catching up after a burst, which is what a job-events mailbox sees if publishers outrun consumers."
|
||||
)
|
||||
print()
|
||||
print("Re-run on NS1: `bash scripts/bench.sh`. Raw logs/CSVs are under `results/<utc>/`.")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Loading…
Add table
Add a link
Reference in a new issue