Add a NATS cluster message-speed bench (throughput and delay).
Some checks are pending
offline / test (push) Waiting to run
Some checks are pending
offline / test (push) Waiting to run
Client LXC 510 against nats-a/b/c at several core and JetStream r=3 loads.
This commit is contained in:
parent
8f0ea8dd31
commit
fef7590c27
34 changed files with 708 additions and 1 deletions
176
packages/verae-nats-cluster/scripts/bench-report.py
Executable file
176
packages/verae-nats-cluster/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())
|
||||
106
packages/verae-nats-cluster/scripts/bench.sh
Executable file
106
packages/verae-nats-cluster/scripts/bench.sh
Executable file
|
|
@ -0,0 +1,106 @@
|
|||
#!/usr/bin/env bash
|
||||
# Message throughput and delay ladder against the 3-node vmbr1 cluster.
|
||||
# Prefers a client that is not a nats-* server (px-worker LXC 510).
|
||||
set -euo pipefail
|
||||
ROOT="$(cd "$(dirname "$0")/.." && pwd)"
|
||||
# shellcheck disable=SC1091
|
||||
. "$ROOT/client.env"
|
||||
export PATH="/usr/sbin:/usr/bin:/bin:/usr/local/bin:$PATH"
|
||||
STAMP="$(date -u +%Y%m%dT%H%M%SZ)"
|
||||
OUT="${BENCH_OUT:-$ROOT/results/$STAMP}"
|
||||
CLIENT_VMID="${CLIENT_VMID:-510}"
|
||||
mkdir -p "$OUT"
|
||||
|
||||
ensure_nats_cli() {
|
||||
local vmid="$1"
|
||||
sudo pct exec "$vmid" -- bash -lc '
|
||||
set -e
|
||||
export DEBIAN_FRONTEND=noninteractive
|
||||
export PATH=/usr/local/bin:/usr/bin:/bin
|
||||
if [[ ! -x /usr/local/bin/nats ]]; then
|
||||
apt-get install -y --no-install-recommends unzip curl ca-certificates >/dev/null
|
||||
curl -fsSL https://github.com/nats-io/natscli/releases/download/v0.1.6/nats-0.1.6-linux-amd64.zip -o /tmp/natscli.zip
|
||||
rm -rf /tmp/natscli && mkdir -p /tmp/natscli
|
||||
unzip -o /tmp/natscli.zip -d /tmp/natscli >/dev/null
|
||||
BIN=$(find /tmp/natscli -type f -name nats | head -1)
|
||||
install -m 0755 "$BIN" /usr/local/bin/nats
|
||||
fi
|
||||
nats --version
|
||||
'
|
||||
}
|
||||
|
||||
run_one() {
|
||||
local name="$1"
|
||||
shift
|
||||
echo "=== $name ===" | tee "$OUT/$name.txt"
|
||||
# nats bench writes csv itself when --csv is a path inside the guest
|
||||
sudo pct exec "$CLIENT_VMID" -- bash -lc "
|
||||
export PATH=/usr/local/bin:/usr/bin:/bin
|
||||
export NATS_URL='$NATS_URL'
|
||||
nats bench --no-progress --csv=/tmp/bench.csv $*
|
||||
" | tee -a "$OUT/$name.txt"
|
||||
sudo pct exec "$CLIENT_VMID" -- cat /tmp/bench.csv >"$OUT/$name.csv" || true
|
||||
}
|
||||
|
||||
echo "client LXC $CLIENT_VMID NATS_URL=$NATS_URL out=$OUT"
|
||||
ensure_nats_cli "$CLIENT_VMID"
|
||||
|
||||
# Core NATS pub/sub — increasing publishers (same 128 B payload)
|
||||
run_one core-1p1s-50k-128 bench.core.a --pub 1 --sub 1 --msgs 50000 --size 128
|
||||
run_one core-4p4s-100k-128 bench.core.b --pub 4 --sub 4 --msgs 100000 --size 128
|
||||
run_one core-8p8s-200k-128 bench.core.c --pub 8 --sub 8 --msgs 200000 --size 128
|
||||
run_one core-4p4s-50k-1k bench.core.d --pub 4 --sub 4 --msgs 50000 --size 1024
|
||||
|
||||
js_rm() {
|
||||
sudo pct exec "$CLIENT_VMID" -- bash -lc "
|
||||
export PATH=/usr/local/bin:/usr/bin:/bin
|
||||
export NATS_URL='$NATS_URL'
|
||||
nats stream rm benchstream --force >/dev/null 2>&1 || true
|
||||
"
|
||||
}
|
||||
# JetStream file store, replicas=3 (matches product streams)
|
||||
js_rm
|
||||
run_one js-1p-20k-128-r3 bench.js.a --js --purge --pub 1 --msgs 20000 --size 128 --replicas 3 --storage file --maxbytes=512MB --stream=benchstream
|
||||
js_rm
|
||||
run_one js-4p-50k-128-r3 bench.js.b --js --purge --pub 4 --msgs 50000 --size 128 --replicas 3 --storage file --maxbytes=512MB --stream=benchstream
|
||||
js_rm
|
||||
run_one js-4p-20k-1k-r3 bench.js.c --js --purge --pub 4 --msgs 20000 --size 1024 --replicas 3 --storage file --maxbytes=512MB --stream=benchstream
|
||||
js_rm
|
||||
run_one js-2p2s-20k-128-r3 bench.js.d --js --purge --pub 2 --sub 2 --msgs 20000 --size 128 --replicas 3 --storage file --maxbytes=512MB --pull --stream=benchstream
|
||||
js_rm
|
||||
|
||||
# Round-trip delay (two connections, through the cluster) at several loads
|
||||
sudo pct exec "$CLIENT_VMID" -- bash -lc "
|
||||
set -e
|
||||
export PATH=/usr/local/bin:/usr/bin:/bin
|
||||
export NATS_URL='$NATS_URL'
|
||||
mkdir -p /tmp/nats-lat
|
||||
cd /tmp/nats-lat
|
||||
if [[ ! -d node_modules/nats ]]; then
|
||||
npm init -y >/dev/null
|
||||
npm install --no-audit --no-fund nats@2 >/dev/null
|
||||
fi
|
||||
" >/dev/null
|
||||
sudo pct push "$CLIENT_VMID" "$ROOT/scripts/latency.mjs" /tmp/nats-lat/latency.mjs || \
|
||||
sudo pct exec "$CLIENT_VMID" -- bash -c 'cat > /tmp/nats-lat/latency.mjs' < "$ROOT/scripts/latency.mjs"
|
||||
lat() {
|
||||
local name="$1" n="$2" sz="$3" p="$4" mode="${5:-flood}"
|
||||
echo "=== $name ===" | tee "$OUT/$name.txt"
|
||||
sudo pct exec "$CLIENT_VMID" -- bash -lc "
|
||||
export PATH=/usr/local/bin:/usr/bin:/bin
|
||||
export NATS_URL='$NATS_URL'
|
||||
cd /tmp/nats-lat
|
||||
node latency.mjs $n $sz $p $mode
|
||||
" | tee -a "$OUT/$name.txt"
|
||||
}
|
||||
# copy latest probe
|
||||
sudo pct exec "$CLIENT_VMID" -- bash -c 'cat > /tmp/nats-lat/latency.mjs' < "$ROOT/scripts/latency.mjs"
|
||||
lat lat-ping-1k-128 1000 128 1 ping
|
||||
lat lat-1p-5k-128 5000 128 1 flood
|
||||
lat lat-4p-10k-128 10000 128 4 flood
|
||||
lat lat-8p-20k-128 20000 128 8 flood
|
||||
lat lat-4p-5k-1k 5000 1024 4 flood
|
||||
|
||||
python3 "$ROOT/scripts/bench-report.py" "$OUT" >"$OUT/BENCH.md"
|
||||
cp "$OUT/BENCH.md" "$ROOT/BENCH.md"
|
||||
echo "wrote $OUT/BENCH.md and $ROOT/BENCH.md"
|
||||
105
packages/verae-nats-cluster/scripts/latency.mjs
Normal file
105
packages/verae-nats-cluster/scripts/latency.mjs
Normal file
|
|
@ -0,0 +1,105 @@
|
|||
#!/usr/bin/env node
|
||||
/**
|
||||
* Pub→sub round trip through the cluster (two connections).
|
||||
* Usage: NATS_URL=... node latency.mjs [count] [payloadBytes] [publishers] [ping|flood]
|
||||
* ping = sequential publish-wait (one-message RTT)
|
||||
* flood = publish the batch then drain (queueing under burst)
|
||||
*/
|
||||
import { connect, headers } from "nats";
|
||||
|
||||
const url = process.env.NATS_URL || "nats://10.10.10.21:4222";
|
||||
const count = Number(process.argv[2] || 5000);
|
||||
const size = Number(process.argv[3] || 128);
|
||||
const pubs = Number(process.argv[4] || 1);
|
||||
const mode = process.argv[5] || "flood";
|
||||
const servers = url.split(",").map((s) => s.trim());
|
||||
const subject = `bench.lat.${process.pid}`;
|
||||
const payload = new Uint8Array(size);
|
||||
|
||||
function pct(sorted, p) {
|
||||
if (!sorted.length) return 0;
|
||||
const i = Math.min(sorted.length - 1, Math.floor((p / 100) * sorted.length));
|
||||
return sorted[i];
|
||||
}
|
||||
|
||||
const samples = [];
|
||||
if (mode === "ping") {
|
||||
const subNc = await connect({ servers, name: "lat-sub" });
|
||||
const pubNc = await connect({ servers, name: "lat-pub" });
|
||||
let resolveOne = null;
|
||||
const sub = subNc.subscribe(subject, { max: count });
|
||||
const consume = (async () => {
|
||||
for await (const m of sub) {
|
||||
const sent = Number(m.headers?.get("t") || 0);
|
||||
samples.push(Number(process.hrtime.bigint() / 1000n) - sent);
|
||||
resolveOne?.();
|
||||
}
|
||||
})();
|
||||
await subNc.flush();
|
||||
for (let i = 0; i < count; i++) {
|
||||
const got = new Promise((r) => {
|
||||
resolveOne = r;
|
||||
});
|
||||
const h = headers();
|
||||
h.set("t", String(process.hrtime.bigint() / 1000n));
|
||||
pubNc.publish(subject, payload, { headers: h });
|
||||
await got;
|
||||
}
|
||||
await consume;
|
||||
await pubNc.close();
|
||||
await subNc.close();
|
||||
} else {
|
||||
const subNc = await connect({ servers, name: "lat-sub" });
|
||||
const sub = subNc.subscribe(subject, { max: count });
|
||||
const done = (async () => {
|
||||
for await (const m of sub) {
|
||||
const sent = Number(m.headers?.get("t") || 0);
|
||||
if (sent) samples.push(Number(process.hrtime.bigint() / 1000n) - sent);
|
||||
}
|
||||
})();
|
||||
await subNc.flush();
|
||||
const per = Math.ceil(count / pubs);
|
||||
const publishers = [];
|
||||
for (let p = 0; p < pubs; p++) {
|
||||
publishers.push(
|
||||
(async () => {
|
||||
const nc = await connect({ servers, name: `lat-pub-${p}` });
|
||||
const n = p === pubs - 1 ? count - per * (pubs - 1) : per;
|
||||
for (let i = 0; i < n; i++) {
|
||||
const h = headers();
|
||||
h.set("t", String(process.hrtime.bigint() / 1000n));
|
||||
nc.publish(subject, payload, { headers: h });
|
||||
}
|
||||
await nc.flush();
|
||||
await nc.close();
|
||||
})(),
|
||||
);
|
||||
}
|
||||
await Promise.all(publishers);
|
||||
await done;
|
||||
await subNc.close();
|
||||
}
|
||||
|
||||
samples.sort((a, b) => a - b);
|
||||
const sum = samples.reduce((a, b) => a + b, 0);
|
||||
const us = (n) => `${(n / 1000).toFixed(3)}ms`;
|
||||
console.log(
|
||||
JSON.stringify({
|
||||
count: samples.length,
|
||||
pubs,
|
||||
size,
|
||||
mode,
|
||||
min_us: samples[0],
|
||||
avg_us: Math.round(sum / samples.length),
|
||||
p50_us: pct(samples, 50),
|
||||
p90_us: pct(samples, 90),
|
||||
p99_us: pct(samples, 99),
|
||||
max_us: samples[samples.length - 1],
|
||||
min: us(samples[0]),
|
||||
avg: us(sum / samples.length),
|
||||
p50: us(pct(samples, 50)),
|
||||
p90: us(pct(samples, 90)),
|
||||
p99: us(pct(samples, 99)),
|
||||
max: us(samples[samples.length - 1]),
|
||||
}),
|
||||
);
|
||||
|
|
@ -5,6 +5,7 @@ ROOT="$(cd "$(dirname "$0")/.." && pwd)"
|
|||
bash -n "$ROOT/scripts/lib-ct.sh"
|
||||
bash -n "$ROOT/scripts/create-cluster.sh"
|
||||
bash -n "$ROOT/scripts/status.sh"
|
||||
bash -n "$ROOT/scripts/bench.sh"
|
||||
grep -q 'host: {{IP}}' "$ROOT/conf/nats.conf.tmpl"
|
||||
grep -qv '0.0.0.0' "$ROOT/conf/nats.conf.tmpl"
|
||||
if [[ ! -d /etc/pve/nodes ]]; then
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue