feat(loadtest): scenario load test harness (#64)

commit: b9978b849f142431714e0965263e07d3679f2247
author: brice · 3d ago · 2026-10-18 12:45 UTC
committer: bnema · 3d ago · 2026-10-18 12:45 UTC
parent: 333c710
signature: SSH UnverifiedSHA256:VOCA5lYARHcbKDZ7GhTV45VmcjWnJ6ZQIBazj17LivA The key is unknown

7 files · +965 -0

A scripts/loadtest/README.md +52 -0

@@ -0,0 +1,52 @@
	1	+	# Load tests
	2	+
	3	+	`loadtest` replays a traffic scenario against an ermite server. It reports throughput and latency per request while holding open the live updates streams (`/-/events`) that real visitors keep.
	4	+
	5	+	## Quick start
	6	+
	7	+	```sh
	8	+	scripts/loadtest/serve.sh # seeded server on :7418, pinned to 1 core
	9	+	go run ./scripts/loadtest -scenario home # closed-loop, 32 clients, 10 s
	10	+	scripts/loadtest/serve.sh stop
	11	+	```
	12	+
	13	+	`serve.sh` builds ermite from the current checkout and seeds five repositories (busy, bursty, steady, private, empty). It pins the server to the last `CPUS` cores (default 1) with `GOMAXPROCS` and `taskset`. A fast workstation would otherwise measure itself, not a small host. Keep the load generator off those cores: `taskset -c 0-15 go run ./scripts/loadtest …`.
	14	+
	15	+	## Modes
	16	+
	17	+	\| Mode \| Flags \| Answers \|
	18	+	\|---\|---\|---\|
	19	+	\| Closed-loop \| `-c 32` \| Maximum throughput: clients send back to back. \|
	20	+	\| Open-loop \| `-rate 1000 -c 64` \| Latency at a target load. Latency counts from when the request was due, so a slow server cannot hide its queue. Requests that find no free client are reported as missed. \|
	21	+
	22	+	Other flags: `-d` duration, `-warmup`, `-streams` (multiplier of the scenario stream counts, `0` for none), `-token` (signed-in pages), `-var name=value`, `-json out.json` (to compare runs), `-url`.
	23	+
	24	+	The run fails (exit 1) on any request error, refused or dropped stream, or missed request.
	25	+
	26	+	Against a loopback server, each stream dials from its own `127.1.x.y` address. The server caps watches per client address (`usecase.MaxWatchersPerClient`) and in total (`usecase.MaxWatchers`).
	27	+
	28	+	## Scenarios
	29	+
	30	+	Built-in scenarios live in `scenarios/*.json`. A file path also works: `-scenario ./my.json`.
	31	+
	32	+	```json
	33	+	{
	34	+	"name": "repo",
	35	+	"requests": [
	36	+	{ "name": "summary", "path": "/{{repo}}", "weight": 4 },
	37	+	{ "name": "commits", "path": "/{{repo}}/commits", "weight": 1 }
	38	+	],
	39	+	"streams": [
	40	+	{ "name": "events", "path": "/-/events?topic=repo:{{repo}}", "count": 50 }
	41	+	]
	42	+	}
	43	+	```
	44	+
	45	+	- `weight` is the share of the requests.
	46	+	- `status` is the expected status code (default 200).
	47	+	- `{{name}}` is replaced by `-var name=value`.
	48	+	- Every request is fetched once before the run; a wrong status stops it early.
	49	+
	50	+	## Reading the figures
	51	+
	52	+	The figures depend on the host. Compare runs made on the same machine with the same `CPUS`, and note the CPU in the report. The report header shows the load generator; the server is whatever `serve.sh` pinned.

@@ -0,0 +1,52 @@
	1	+	# Load tests
	2	+
	3	+	`loadtest` replays a traffic scenario against an ermite server. It reports throughput and latency per request while holding open the live updates streams (`/-/events`) that real visitors keep.
	4	+
	5	+	## Quick start
	6	+
	7	+	```sh
	8	+	scripts/loadtest/serve.sh # seeded server on :7418, pinned to 1 core
	9	+	go run ./scripts/loadtest -scenario home # closed-loop, 32 clients, 10 s
	10	+	scripts/loadtest/serve.sh stop
	11	+	```
	12	+
	13	+	`serve.sh` builds ermite from the current checkout and seeds five repositories (busy, bursty, steady, private, empty). It pins the server to the last `CPUS` cores (default 1) with `GOMAXPROCS` and `taskset`. A fast workstation would otherwise measure itself, not a small host. Keep the load generator off those cores: `taskset -c 0-15 go run ./scripts/loadtest …`.
	14	+
	15	+	## Modes
	16	+
	17	+	\| Mode \| Flags \| Answers \|
	18	+	\|---\|---\|---\|
	19	+	\| Closed-loop \| `-c 32` \| Maximum throughput: clients send back to back. \|
	20	+	\| Open-loop \| `-rate 1000 -c 64` \| Latency at a target load. Latency counts from when the request was due, so a slow server cannot hide its queue. Requests that find no free client are reported as missed. \|
	21	+
	22	+	Other flags: `-d` duration, `-warmup`, `-streams` (multiplier of the scenario stream counts, `0` for none), `-token` (signed-in pages), `-var name=value`, `-json out.json` (to compare runs), `-url`.
	23	+
	24	+	The run fails (exit 1) on any request error, refused or dropped stream, or missed request.
	25	+
	26	+	Against a loopback server, each stream dials from its own `127.1.x.y` address. The server caps watches per client address (`usecase.MaxWatchersPerClient`) and in total (`usecase.MaxWatchers`).
	27	+
	28	+	## Scenarios
	29	+
	30	+	Built-in scenarios live in `scenarios/*.json`. A file path also works: `-scenario ./my.json`.
	31	+
	32	+	```json
	33	+	{
	34	+	"name": "repo",
	35	+	"requests": [
	36	+	{ "name": "summary", "path": "/{{repo}}", "weight": 4 },
	37	+	{ "name": "commits", "path": "/{{repo}}/commits", "weight": 1 }
	38	+	],
	39	+	"streams": [
	40	+	{ "name": "events", "path": "/-/events?topic=repo:{{repo}}", "count": 50 }
	41	+	]
	42	+	}
	43	+	```
	44	+
	45	+	- `weight` is the share of the requests.
	46	+	- `status` is the expected status code (default 200).
	47	+	- `{{name}}` is replaced by `-var name=value`.
	48	+	- Every request is fetched once before the run; a wrong status stops it early.
	49	+
	50	+	## Reading the figures
	51	+
	52	+	The figures depend on the host. Compare runs made on the same machine with the same `CPUS`, and note the CPU in the report. The report header shows the load generator; the server is whatever `serve.sh` pinned.

A scripts/loadtest/main.go +106 -0

@@ -0,0 +1,106 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	// loadtest replays a traffic scenario against an ermite server and reports
	4	+	// throughput and latency per request, while holding open the live updates
	5	+	// streams that the pages of real visitors keep.
	6	+	//
	7	+	// go run ./scripts/loadtest -url http://127.0.0.1:7418 -scenario home
	8	+	// go run ./scripts/loadtest -scenario repo -var repo=neferwl -rate 500
	9	+	//
	10	+	// Without -rate it runs closed-loop: -c clients send requests back to back,
	11	+	// which measures the maximum throughput. With -rate it runs open-loop: it
	12	+	// sends that many requests per second whatever the latency, which measures
	13	+	// the latency at a target load; a request that finds no free client is
	14	+	// counted as missed. See scripts/loadtest/README.md.
	15	+	package main
	16	+
	17	+	import (
	18	+	"context"
	19	+	"encoding/json"
	20	+	"flag"
	21	+	"fmt"
	22	+	"math"
	23	+	"os"
	24	+	"os/signal"
	25	+	"strings"
	26	+	"time"
	27	+	)
	28	+
	29	+	func (c config) validate() error {
	30	+	switch {
	31	+	case c.clients < 1:
	32	+	return fmt.Errorf("-c must be at least 1, got %d", c.clients)
	33	+	case c.duration <= 0:
	34	+	return fmt.Errorf("-d must be positive, got %v", c.duration)
	35	+	case c.warmup < 0:
	36	+	return fmt.Errorf("-warmup must not be negative, got %v", c.warmup)
	37	+	case !(c.streams >= 0) \|\| math.IsInf(c.streams, 0):
	38	+	return fmt.Errorf("-streams must be a finite number >= 0, got %v", c.streams)
	39	+	case !(c.rate >= 0) \|\| math.IsInf(c.rate, 0):
	40	+	return fmt.Errorf("-rate must be a finite number >= 0, got %v", c.rate)
	41	+	case c.rate > 1e6:
	42	+	return fmt.Errorf("-rate is at most 1000000, got %v", c.rate)
	43	+	}
	44	+	return nil
	45	+	}
	46	+
	47	+	type varsFlag map[string]string
	48	+
	49	+	func (v varsFlag) String() string { return fmt.Sprint(map[string]string(v)) }
	50	+
	51	+	func (v varsFlag) Set(s string) error {
	52	+	k, val, ok := strings.Cut(s, "=")
	53	+	if !ok \|\| k == "" {
	54	+	return fmt.Errorf("want name=value, got %q", s)
	55	+	}
	56	+	v[k] = val
	57	+	return nil
	58	+	}
	59	+
	60	+	func main() {
	61	+	cfg := config{vars: varsFlag{}}
	62	+	flag.StringVar(&cfg.base, "url", "http://127.0.0.1:7418", "base URL of the server")
	63	+	flag.StringVar(&cfg.scenario, "scenario", "home", "built-in scenario name, or path to a .json file")
	64	+	flag.Var(cfg.vars, "var", "scenario variable name=value, repeatable")
	65	+	flag.IntVar(&cfg.clients, "c", 32, "concurrent clients")
	66	+	flag.Float64Var(&cfg.rate, "rate", 0, "requests per second (open-loop); 0 runs closed-loop")
	67	+	flag.DurationVar(&cfg.duration, "d", 10*time.Second, "measured duration")
	68	+	flag.DurationVar(&cfg.warmup, "warmup", 2*time.Second, "unmeasured warm-up before the run")
	69	+	flag.Float64Var(&cfg.streams, "streams", 1, "multiplier of the stream counts of the scenario; 0 opens none")
	70	+	flag.StringVar(&cfg.token, "token", "", "bearer token sent with every request (signed-in pages)")
	71	+	flag.BoolVar(&cfg.gzip, "gzip", true, "ask for compressed responses, as browsers do")
	72	+	flag.StringVar(&cfg.jsonOut, "json", "", "also write the report as JSON to this file")
	73	+	flag.Parse()
	74	+
	75	+	if err := cfg.validate(); err != nil {
	76	+	fmt.Fprintln(os.Stderr, "loadtest:", err)
	77	+	os.Exit(2)
	78	+	}
	79	+	sc, err := loadScenario(cfg.scenario, cfg.vars)
	80	+	if err != nil {
	81	+	fmt.Fprintln(os.Stderr, "loadtest:", err)
	82	+	os.Exit(2)
	83	+	}
	84	+	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt)
	85	+	defer stop()
	86	+
	87	+	rep, err := run(ctx, cfg, sc)
	88	+	if err != nil {
	89	+	fmt.Fprintln(os.Stderr, "loadtest:", err)
	90	+	os.Exit(1)
	91	+	}
	92	+	if err := rep.print(os.Stdout); err != nil {
	93	+	fmt.Fprintln(os.Stderr, "loadtest:", err)
	94	+	os.Exit(1)
	95	+	}
	96	+	if cfg.jsonOut != "" {
	97	+	data, _ := json.MarshalIndent(rep, "", " ")
	98	+	if err := os.WriteFile(cfg.jsonOut, append(data, '\n'), 0o644); err != nil {
	99	+	fmt.Fprintln(os.Stderr, "loadtest:", err)
	100	+	os.Exit(1)
	101	+	}
	102	+	}
	103	+	if rep.Failed() {
	104	+	os.Exit(1)
	105	+	}
	106	+	}

@@ -0,0 +1,106 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	// loadtest replays a traffic scenario against an ermite server and reports
	4	+	// throughput and latency per request, while holding open the live updates
	5	+	// streams that the pages of real visitors keep.
	6	+	//
	7	+	// go run ./scripts/loadtest -url http://127.0.0.1:7418 -scenario home
	8	+	// go run ./scripts/loadtest -scenario repo -var repo=neferwl -rate 500
	9	+	//
	10	+	// Without -rate it runs closed-loop: -c clients send requests back to back,
	11	+	// which measures the maximum throughput. With -rate it runs open-loop: it
	12	+	// sends that many requests per second whatever the latency, which measures
	13	+	// the latency at a target load; a request that finds no free client is
	14	+	// counted as missed. See scripts/loadtest/README.md.
	15	+	package main
	16	+
	17	+	import (
	18	+	"context"
	19	+	"encoding/json"
	20	+	"flag"
	21	+	"fmt"
	22	+	"math"
	23	+	"os"
	24	+	"os/signal"
	25	+	"strings"
	26	+	"time"
	27	+	)
	28	+
	29	+	func (c config) validate() error {
	30	+	switch {
	31	+	case c.clients < 1:
	32	+	return fmt.Errorf("-c must be at least 1, got %d", c.clients)
	33	+	case c.duration <= 0:
	34	+	return fmt.Errorf("-d must be positive, got %v", c.duration)
	35	+	case c.warmup < 0:
	36	+	return fmt.Errorf("-warmup must not be negative, got %v", c.warmup)
	37	+	case !(c.streams >= 0) \|\| math.IsInf(c.streams, 0):
	38	+	return fmt.Errorf("-streams must be a finite number >= 0, got %v", c.streams)
	39	+	case !(c.rate >= 0) \|\| math.IsInf(c.rate, 0):
	40	+	return fmt.Errorf("-rate must be a finite number >= 0, got %v", c.rate)
	41	+	case c.rate > 1e6:
	42	+	return fmt.Errorf("-rate is at most 1000000, got %v", c.rate)
	43	+	}
	44	+	return nil
	45	+	}
	46	+
	47	+	type varsFlag map[string]string
	48	+
	49	+	func (v varsFlag) String() string { return fmt.Sprint(map[string]string(v)) }
	50	+
	51	+	func (v varsFlag) Set(s string) error {
	52	+	k, val, ok := strings.Cut(s, "=")
	53	+	if !ok \|\| k == "" {
	54	+	return fmt.Errorf("want name=value, got %q", s)
	55	+	}
	56	+	v[k] = val
	57	+	return nil
	58	+	}
	59	+
	60	+	func main() {
	61	+	cfg := config{vars: varsFlag{}}
	62	+	flag.StringVar(&cfg.base, "url", "http://127.0.0.1:7418", "base URL of the server")
	63	+	flag.StringVar(&cfg.scenario, "scenario", "home", "built-in scenario name, or path to a .json file")
	64	+	flag.Var(cfg.vars, "var", "scenario variable name=value, repeatable")
	65	+	flag.IntVar(&cfg.clients, "c", 32, "concurrent clients")
	66	+	flag.Float64Var(&cfg.rate, "rate", 0, "requests per second (open-loop); 0 runs closed-loop")
	67	+	flag.DurationVar(&cfg.duration, "d", 10*time.Second, "measured duration")
	68	+	flag.DurationVar(&cfg.warmup, "warmup", 2*time.Second, "unmeasured warm-up before the run")
	69	+	flag.Float64Var(&cfg.streams, "streams", 1, "multiplier of the stream counts of the scenario; 0 opens none")
	70	+	flag.StringVar(&cfg.token, "token", "", "bearer token sent with every request (signed-in pages)")
	71	+	flag.BoolVar(&cfg.gzip, "gzip", true, "ask for compressed responses, as browsers do")
	72	+	flag.StringVar(&cfg.jsonOut, "json", "", "also write the report as JSON to this file")
	73	+	flag.Parse()
	74	+
	75	+	if err := cfg.validate(); err != nil {
	76	+	fmt.Fprintln(os.Stderr, "loadtest:", err)
	77	+	os.Exit(2)
	78	+	}
	79	+	sc, err := loadScenario(cfg.scenario, cfg.vars)
	80	+	if err != nil {
	81	+	fmt.Fprintln(os.Stderr, "loadtest:", err)
	82	+	os.Exit(2)
	83	+	}
	84	+	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt)
	85	+	defer stop()
	86	+
	87	+	rep, err := run(ctx, cfg, sc)
	88	+	if err != nil {
	89	+	fmt.Fprintln(os.Stderr, "loadtest:", err)
	90	+	os.Exit(1)
	91	+	}
	92	+	if err := rep.print(os.Stdout); err != nil {
	93	+	fmt.Fprintln(os.Stderr, "loadtest:", err)
	94	+	os.Exit(1)
	95	+	}
	96	+	if cfg.jsonOut != "" {
	97	+	data, _ := json.MarshalIndent(rep, "", " ")
	98	+	if err := os.WriteFile(cfg.jsonOut, append(data, '\n'), 0o644); err != nil {
	99	+	fmt.Fprintln(os.Stderr, "loadtest:", err)
	100	+	os.Exit(1)
	101	+	}
	102	+	}
	103	+	if rep.Failed() {
	104	+	os.Exit(1)
	105	+	}
	106	+	}

A scripts/loadtest/report.go +195 -0

@@ -0,0 +1,195 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package main
	4	+
	5	+	import (
	6	+	"bytes"
	7	+	"fmt"
	8	+	"io"
	9	+	"maps"
	10	+	"runtime"
	11	+	"slices"
	12	+	"text/tabwriter"
	13	+	"time"
	14	+	)
	15	+
	16	+	// Report is the result of a run, printed as a table and optionally written as
	17	+	// JSON to compare runs.
	18	+	type Report struct {
	19	+	Scenario string `json:"scenario"`
	20	+	URL string `json:"url"`
	21	+	Started time.Time `json:"started"`
	22	+	Mode string `json:"mode"` // "closed-loop" or "open-loop"
	23	+	Clients int `json:"clients"`
	24	+	Rate float64 `json:"target_rps,omitempty"`
	25	+	Seconds float64 `json:"seconds"`
	26	+	Client string `json:"client"` // the load generator, to tell runs apart
	27	+	Total ReqReport `json:"total"`
	28	+	// Missed counts the open-loop requests that found no free client: the
	29	+	// server could not keep up with the target rate.
	30	+	Missed int64 `json:"missed"`
	31	+	Requests []ReqReport `json:"requests"`
	32	+	Streams []StreamReport `json:"streams"`
	33	+	}
	34	+
	35	+	type ReqReport struct {
	36	+	Name string `json:"name"`
	37	+	Path string `json:"path,omitempty"`
	38	+	Count int `json:"count"`
	39	+	RPS float64 `json:"rps"`
	40	+	Errors map[string]int `json:"errors,omitempty"`
	41	+	Bytes int64 `json:"avg_bytes"`
	42	+	Latency Latency `json:"latency_ms"`
	43	+	}
	44	+
	45	+	type StreamReport struct {
	46	+	Name string `json:"name"`
	47	+	Wanted int `json:"wanted"`
	48	+	Opened int `json:"opened"`
	49	+	Refused int `json:"refused"`
	50	+	Dropped int `json:"dropped"`
	51	+	Events int `json:"events"`
	52	+	Connect Latency `json:"connect_ms"`
	53	+	}
	54	+
	55	+	type Latency struct {
	56	+	P50 float64 `json:"p50"`
	57	+	P90 float64 `json:"p90"`
	58	+	P99 float64 `json:"p99"`
	59	+	Max float64 `json:"max"`
	60	+	}
	61	+
	62	+	func summarize(d []time.Duration) Latency {
	63	+	if len(d) == 0 {
	64	+	return Latency{}
	65	+	}
	66	+	s := slices.Clone(d)
	67	+	slices.Sort(s)
	68	+	q := func(p float64) float64 {
	69	+	return float64(s[int(float64(len(s)-1)*p)]) / float64(time.Millisecond)
	70	+	}
	71	+	return Latency{P50: q(.5), P90: q(.9), P99: q(.99), Max: q(1)}
	72	+	}
	73	+
	74	+	func newReport(cfg config, sc Scenario, samples []sample, missed int64, elapsed time.Duration, streams []StreamReport) Report {
	75	+	r := &Report{
	76	+	Scenario: sc.Name, URL: cfg.base, Started: time.Now().Add(-elapsed).UTC(),
	77	+	Mode: "closed-loop", Clients: cfg.clients, Seconds: elapsed.Seconds(),
	78	+	Client: fmt.Sprintf("%s/%s, %d CPUs", runtime.GOOS, runtime.GOARCH, runtime.NumCPU()),
	79	+	Missed: missed,
	80	+	Streams: streams,
	81	+	}
	82	+	if cfg.rate > 0 {
	83	+	r.Mode, r.Rate = "open-loop", cfg.rate
	84	+	}
	85	+	sum := func(name, path string, ss []sample) ReqReport {
	86	+	rr := ReqReport{Name: name, Path: path, Count: len(ss), RPS: float64(len(ss)) / elapsed.Seconds()}
	87	+	lat := make([]time.Duration, 0, len(ss))
	88	+	var bytes int64
	89	+	for _, s := range ss {
	90	+	lat = append(lat, s.latency)
	91	+	bytes += s.bytes
	92	+	if s.err != "" {
	93	+	if rr.Errors == nil {
	94	+	rr.Errors = map[string]int{}
	95	+	}
	96	+	rr.Errors[s.err]++
	97	+	}
	98	+	}
	99	+	if len(ss) > 0 {
	100	+	rr.Bytes = bytes / int64(len(ss))
	101	+	}
	102	+	rr.Latency = summarize(lat)
	103	+	return rr
	104	+	}
	105	+	byReq := make([][]sample, len(sc.Requests))
	106	+	for _, s := range samples {
	107	+	byReq[s.req] = append(byReq[s.req], s)
	108	+	}
	109	+	for i, q := range sc.Requests {
	110	+	r.Requests = append(r.Requests, sum(q.Name, q.Path, byReq[i]))
	111	+	}
	112	+	r.Total = sum("total", "", samples)
	113	+	return r
	114	+	}
	115	+
	116	+	// Failed is true when a request failed, a stream did not open or was
	117	+	// dropped, or the server missed the target rate.
	118	+	func (r *Report) Failed() bool {
	119	+	if len(r.Total.Errors) > 0 \|\| r.Missed > 0 {
	120	+	return true
	121	+	}
	122	+	for _, s := range r.Streams {
	123	+	if s.Opened < s.Wanted \|\| s.Refused > 0 \|\| s.Dropped > 0 {
	124	+	return true
	125	+	}
	126	+	}
	127	+	return false
	128	+	}
	129	+
	130	+	// print writes the report as text; it is built in memory (writes to a
	131	+	// bytes.Buffer cannot fail) and written once.
	132	+	func (r *Report) print(out io.Writer) error {
	133	+	w := &bytes.Buffer{}
	134	+	fmt.Fprintf(w, "scenario %s on %s, %s, %d clients", r.Scenario, r.URL, r.Mode, r.Clients)
	135	+	if r.Rate > 0 {
	136	+	fmt.Fprintf(w, ", target %.0f req/s", r.Rate)
	137	+	}
	138	+	fmt.Fprintf(w, ", %.1fs (load generator: %s)\n\n", r.Seconds, r.Client)
	139	+
	140	+	tw := &bytes.Buffer{}
	141	+	fmt.Fprintln(tw, "request\treq/s\tp50 ms\tp90 ms\tp99 ms\tmax ms\tavg KB\terrors\t")
	142	+	row := func(q ReqReport) {
	143	+	fmt.Fprintf(tw, "%s\t%.0f\t%.2f\t%.2f\t%.2f\t%.1f\t%.1f\t%d\t\n", q.Name, q.RPS,
	144	+	q.Latency.P50, q.Latency.P90, q.Latency.P99, q.Latency.Max, float64(q.Bytes)/1024, errCount(q.Errors))
	145	+	}
	146	+	for _, q := range r.Requests {
	147	+	row(q)
	148	+	}
	149	+	if len(r.Requests) > 1 {
	150	+	row(r.Total)
	151	+	}
	152	+	table(w, tw)
	153	+
	154	+	for _, q := range append(r.Requests, r.Total) {
	155	+	for _, e := range slices.Sorted(maps.Keys(q.Errors)) {
	156	+	if q.Name != "total" \|\| len(r.Requests) == 1 {
	157	+	fmt.Fprintf(w, " %s: %d × %s\n", q.Name, q.Errors[e], e)
	158	+	}
	159	+	}
	160	+	}
	161	+	if r.Missed > 0 {
	162	+	fmt.Fprintf(w, "\nmissed %d requests: the server did not keep up with %.0f req/s\n", r.Missed, r.Rate)
	163	+	}
	164	+	if len(r.Streams) > 0 {
	165	+	w.WriteByte('\n')
	166	+	tw.Reset()
	167	+	fmt.Fprintln(tw, "stream\topen\trefused\tdropped\tevents\tconnect p99 ms\t")
	168	+	for _, s := range r.Streams {
	169	+	fmt.Fprintf(tw, "%s\t%d/%d\t%d\t%d\t%d\t%.2f\t\n", s.Name, s.Opened, s.Wanted, s.Refused, s.Dropped, s.Events, s.Connect.P99)
	170	+	}
	171	+	table(w, tw)
	172	+	}
	173	+	verdict := "PASS"
	174	+	if r.Failed() {
	175	+	verdict = "FAIL"
	176	+	}
	177	+	fmt.Fprintf(w, "\n%s\n", verdict)
	178	+	_, err := out.Write(w.Bytes())
	179	+	return err
	180	+	}
	181	+
	182	+	// table aligns the tab-separated rows of src into dst.
	183	+	func table(dst, src *bytes.Buffer) {
	184	+	tw := tabwriter.NewWriter(dst, 0, 0, 2, ' ', tabwriter.AlignRight)
	185	+	_, _ = tw.Write(src.Bytes())
	186	+	_ = tw.Flush() // cannot fail: dst is a bytes.Buffer
	187	+	}
	188	+
	189	+	func errCount(m map[string]int) int {
	190	+	n := 0
	191	+	for _, v := range m {
	192	+	n += v
	193	+	}
	194	+	return n
	195	+	}

@@ -0,0 +1,195 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package main
	4	+
	5	+	import (
	6	+	"bytes"
	7	+	"fmt"
	8	+	"io"
	9	+	"maps"
	10	+	"runtime"
	11	+	"slices"
	12	+	"text/tabwriter"
	13	+	"time"
	14	+	)
	15	+
	16	+	// Report is the result of a run, printed as a table and optionally written as
	17	+	// JSON to compare runs.
	18	+	type Report struct {
	19	+	Scenario string `json:"scenario"`
	20	+	URL string `json:"url"`
	21	+	Started time.Time `json:"started"`
	22	+	Mode string `json:"mode"` // "closed-loop" or "open-loop"
	23	+	Clients int `json:"clients"`
	24	+	Rate float64 `json:"target_rps,omitempty"`
	25	+	Seconds float64 `json:"seconds"`
	26	+	Client string `json:"client"` // the load generator, to tell runs apart
	27	+	Total ReqReport `json:"total"`
	28	+	// Missed counts the open-loop requests that found no free client: the
	29	+	// server could not keep up with the target rate.
	30	+	Missed int64 `json:"missed"`
	31	+	Requests []ReqReport `json:"requests"`
	32	+	Streams []StreamReport `json:"streams"`
	33	+	}
	34	+
	35	+	type ReqReport struct {
	36	+	Name string `json:"name"`
	37	+	Path string `json:"path,omitempty"`
	38	+	Count int `json:"count"`
	39	+	RPS float64 `json:"rps"`
	40	+	Errors map[string]int `json:"errors,omitempty"`
	41	+	Bytes int64 `json:"avg_bytes"`
	42	+	Latency Latency `json:"latency_ms"`
	43	+	}
	44	+
	45	+	type StreamReport struct {
	46	+	Name string `json:"name"`
	47	+	Wanted int `json:"wanted"`
	48	+	Opened int `json:"opened"`
	49	+	Refused int `json:"refused"`
	50	+	Dropped int `json:"dropped"`
	51	+	Events int `json:"events"`
	52	+	Connect Latency `json:"connect_ms"`
	53	+	}
	54	+
	55	+	type Latency struct {
	56	+	P50 float64 `json:"p50"`
	57	+	P90 float64 `json:"p90"`
	58	+	P99 float64 `json:"p99"`
	59	+	Max float64 `json:"max"`
	60	+	}
	61	+
	62	+	func summarize(d []time.Duration) Latency {
	63	+	if len(d) == 0 {
	64	+	return Latency{}
	65	+	}
	66	+	s := slices.Clone(d)
	67	+	slices.Sort(s)
	68	+	q := func(p float64) float64 {
	69	+	return float64(s[int(float64(len(s)-1)*p)]) / float64(time.Millisecond)
	70	+	}
	71	+	return Latency{P50: q(.5), P90: q(.9), P99: q(.99), Max: q(1)}
	72	+	}
	73	+
	74	+	func newReport(cfg config, sc Scenario, samples []sample, missed int64, elapsed time.Duration, streams []StreamReport) Report {
	75	+	r := &Report{
	76	+	Scenario: sc.Name, URL: cfg.base, Started: time.Now().Add(-elapsed).UTC(),
	77	+	Mode: "closed-loop", Clients: cfg.clients, Seconds: elapsed.Seconds(),
	78	+	Client: fmt.Sprintf("%s/%s, %d CPUs", runtime.GOOS, runtime.GOARCH, runtime.NumCPU()),
	79	+	Missed: missed,
	80	+	Streams: streams,
	81	+	}
	82	+	if cfg.rate > 0 {
	83	+	r.Mode, r.Rate = "open-loop", cfg.rate
	84	+	}
	85	+	sum := func(name, path string, ss []sample) ReqReport {
	86	+	rr := ReqReport{Name: name, Path: path, Count: len(ss), RPS: float64(len(ss)) / elapsed.Seconds()}
	87	+	lat := make([]time.Duration, 0, len(ss))
	88	+	var bytes int64
	89	+	for _, s := range ss {
	90	+	lat = append(lat, s.latency)
	91	+	bytes += s.bytes
	92	+	if s.err != "" {
	93	+	if rr.Errors == nil {
	94	+	rr.Errors = map[string]int{}
	95	+	}
	96	+	rr.Errors[s.err]++
	97	+	}
	98	+	}
	99	+	if len(ss) > 0 {
	100	+	rr.Bytes = bytes / int64(len(ss))
	101	+	}
	102	+	rr.Latency = summarize(lat)
	103	+	return rr
	104	+	}
	105	+	byReq := make([][]sample, len(sc.Requests))
	106	+	for _, s := range samples {
	107	+	byReq[s.req] = append(byReq[s.req], s)
	108	+	}
	109	+	for i, q := range sc.Requests {
	110	+	r.Requests = append(r.Requests, sum(q.Name, q.Path, byReq[i]))
	111	+	}
	112	+	r.Total = sum("total", "", samples)
	113	+	return r
	114	+	}
	115	+
	116	+	// Failed is true when a request failed, a stream did not open or was
	117	+	// dropped, or the server missed the target rate.
	118	+	func (r *Report) Failed() bool {
	119	+	if len(r.Total.Errors) > 0 \|\| r.Missed > 0 {
	120	+	return true
	121	+	}
	122	+	for _, s := range r.Streams {
	123	+	if s.Opened < s.Wanted \|\| s.Refused > 0 \|\| s.Dropped > 0 {
	124	+	return true
	125	+	}
	126	+	}
	127	+	return false
	128	+	}
	129	+
	130	+	// print writes the report as text; it is built in memory (writes to a
	131	+	// bytes.Buffer cannot fail) and written once.
	132	+	func (r *Report) print(out io.Writer) error {
	133	+	w := &bytes.Buffer{}
	134	+	fmt.Fprintf(w, "scenario %s on %s, %s, %d clients", r.Scenario, r.URL, r.Mode, r.Clients)
	135	+	if r.Rate > 0 {
	136	+	fmt.Fprintf(w, ", target %.0f req/s", r.Rate)
	137	+	}
	138	+	fmt.Fprintf(w, ", %.1fs (load generator: %s)\n\n", r.Seconds, r.Client)
	139	+
	140	+	tw := &bytes.Buffer{}
	141	+	fmt.Fprintln(tw, "request\treq/s\tp50 ms\tp90 ms\tp99 ms\tmax ms\tavg KB\terrors\t")
	142	+	row := func(q ReqReport) {
	143	+	fmt.Fprintf(tw, "%s\t%.0f\t%.2f\t%.2f\t%.2f\t%.1f\t%.1f\t%d\t\n", q.Name, q.RPS,
	144	+	q.Latency.P50, q.Latency.P90, q.Latency.P99, q.Latency.Max, float64(q.Bytes)/1024, errCount(q.Errors))
	145	+	}
	146	+	for _, q := range r.Requests {
	147	+	row(q)
	148	+	}
	149	+	if len(r.Requests) > 1 {
	150	+	row(r.Total)
	151	+	}
	152	+	table(w, tw)
	153	+
	154	+	for _, q := range append(r.Requests, r.Total) {
	155	+	for _, e := range slices.Sorted(maps.Keys(q.Errors)) {
	156	+	if q.Name != "total" \|\| len(r.Requests) == 1 {
	157	+	fmt.Fprintf(w, " %s: %d × %s\n", q.Name, q.Errors[e], e)
	158	+	}
	159	+	}
	160	+	}
	161	+	if r.Missed > 0 {
	162	+	fmt.Fprintf(w, "\nmissed %d requests: the server did not keep up with %.0f req/s\n", r.Missed, r.Rate)
	163	+	}
	164	+	if len(r.Streams) > 0 {
	165	+	w.WriteByte('\n')
	166	+	tw.Reset()
	167	+	fmt.Fprintln(tw, "stream\topen\trefused\tdropped\tevents\tconnect p99 ms\t")
	168	+	for _, s := range r.Streams {
	169	+	fmt.Fprintf(tw, "%s\t%d/%d\t%d\t%d\t%d\t%.2f\t\n", s.Name, s.Opened, s.Wanted, s.Refused, s.Dropped, s.Events, s.Connect.P99)
	170	+	}
	171	+	table(w, tw)
	172	+	}
	173	+	verdict := "PASS"
	174	+	if r.Failed() {
	175	+	verdict = "FAIL"
	176	+	}
	177	+	fmt.Fprintf(w, "\n%s\n", verdict)
	178	+	_, err := out.Write(w.Bytes())
	179	+	return err
	180	+	}
	181	+
	182	+	// table aligns the tab-separated rows of src into dst.
	183	+	func table(dst, src *bytes.Buffer) {
	184	+	tw := tabwriter.NewWriter(dst, 0, 0, 2, ' ', tabwriter.AlignRight)
	185	+	_, _ = tw.Write(src.Bytes())
	186	+	_ = tw.Flush() // cannot fail: dst is a bytes.Buffer
	187	+	}
	188	+
	189	+	func errCount(m map[string]int) int {
	190	+	n := 0
	191	+	for _, v := range m {
	192	+	n += v
	193	+	}
	194	+	return n
	195	+	}

A scripts/loadtest/run.go +370 -0

@@ -0,0 +1,370 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package main
	4	+
	5	+	import (
	6	+	"bufio"
	7	+	"context"
	8	+	"errors"
	9	+	"fmt"
	10	+	"io"
	11	+	"math/rand/v2"
	12	+	"net"
	13	+	"net/http"
	14	+	"net/netip"
	15	+	"net/url"
	16	+	"strings"
	17	+	"sync"
	18	+	"sync/atomic"
	19	+	"time"
	20	+	)
	21	+
	22	+	type config struct {
	23	+	base string
	24	+	scenario string
	25	+	vars varsFlag
	26	+	clients int
	27	+	rate float64
	28	+	duration time.Duration
	29	+	warmup time.Duration
	30	+	streams float64
	31	+	token string
	32	+	gzip bool
	33	+	jsonOut string
	34	+	}
	35	+
	36	+	// sample is one finished request.
	37	+	type sample struct {
	38	+	req int
	39	+	latency time.Duration
	40	+	bytes int64
	41	+	err string // "" when the request got the expected status
	42	+	}
	43	+
	44	+	// job is one request to send; at is when it was due, so that open-loop
	45	+	// latencies include the time spent waiting for a client (no coordinated
	46	+	// omission).
	47	+	type job struct {
	48	+	req int
	49	+	at time.Time
	50	+	}
	51	+
	52	+	func run(ctx context.Context, cfg config, sc Scenario) (Report, error) {
	53	+	base := strings.TrimRight(cfg.base, "/")
	54	+	tr := &http.Transport{
	55	+	MaxIdleConns: cfg.clients * 2,
	56	+	MaxIdleConnsPerHost: cfg.clients * 2,
	57	+	DisableCompression: true, // the Accept-Encoding header is set by hand
	58	+	IdleConnTimeout: 90 * time.Second,
	59	+	}
	60	+	client := &http.Client{Transport: tr, Timeout: 30 * time.Second}
	61	+	defer tr.CloseIdleConnections()
	62	+
	63	+	if err := preflight(ctx, client, cfg, base, sc); err != nil {
	64	+	return nil, err
	65	+	}
	66	+
	67	+	// Streams are opened first and stay open for the whole run.
	68	+	sctx, scancel := context.WithCancel(ctx)
	69	+	defer scancel()
	70	+	streamStats := openStreams(sctx, cfg, base, sc)
	71	+
	72	+	if cfg.warmup > 0 {
	73	+	fire(ctx, client, cfg, base, sc, cfg.warmup, nil)
	74	+	}
	75	+	var mu sync.Mutex
	76	+	var samples []sample
	77	+	collect := func(s sample) {
	78	+	mu.Lock()
	79	+	samples = append(samples, s)
	80	+	mu.Unlock()
	81	+	}
	82	+	start := time.Now()
	83	+	missed := fire(ctx, client, cfg, base, sc, cfg.duration, collect)
	84	+	elapsed := time.Since(start)
	85	+	scancel()
	86	+	streams := streamStats.wait()
	87	+	if err := ctx.Err(); err != nil {
	88	+	return nil, fmt.Errorf("interrupted after %v: %w", elapsed.Round(time.Millisecond), err)
	89	+	}
	90	+
	91	+	return newReport(cfg, sc, samples, missed, elapsed, streams), nil
	92	+	}
	93	+
	94	+	// preflight fetches every request once and fails fast on a wrong status, so
	95	+	// that a typo in a scenario does not turn into a run of errors.
	96	+	func preflight(ctx context.Context, client http.Client, cfg config, base string, sc Scenario) error {
	97	+	var errs []string
	98	+	for _, r := range sc.Requests {
	99	+	s := do(ctx, client, cfg, base, r)
	100	+	if s.err != "" {
	101	+	errs = append(errs, fmt.Sprintf("%s (%s): %s", r.Name, r.Path, s.err))
	102	+	}
	103	+	}
	104	+	if len(errs) > 0 {
	105	+	return errors.New("preflight failed:\n " + strings.Join(errs, "\n "))
	106	+	}
	107	+	return nil
	108	+	}
	109	+
	110	+	// fire sends requests for d, closed-loop or open-loop, and hands each result
	111	+	// to collect (nil during the warm-up). It returns the open-loop requests that
	112	+	// found no free client.
	113	+	//
	114	+	// The deadline only stops new requests: those in flight or queued when it
	115	+	// passes still run to the end (bounded by the client timeout) and are
	116	+	// counted, so that slow responses are not dropped from the figures. Only an
	117	+	// interruption of ctx cancels them.
	118	+	func fire(ctx context.Context, client http.Client, cfg config, base string, sc Scenario, d time.Duration, collect func(sample)) int64 {
	119	+	sched, cancel := context.WithTimeout(ctx, d)
	120	+	defer cancel()
	121	+	pk := newPicker(sc.Requests)
	122	+	var wg sync.WaitGroup
	123	+
	124	+	send := func(j job) {
	125	+	s := do(ctx, client, cfg, base, sc.Requests[j.req])
	126	+	if ctx.Err() != nil {
	127	+	return // interrupted: the run reports no figures
	128	+	}
	129	+	s.req = j.req
	130	+	if !j.at.IsZero() {
	131	+	s.latency = time.Since(j.at)
	132	+	}
	133	+	if collect != nil {
	134	+	collect(s)
	135	+	}
	136	+	}
	137	+
	138	+	if cfg.rate <= 0 {
	139	+	for i := range cfg.clients {
	140	+	rng := rand.New(rand.NewPCG(uint64(i), uint64(time.Now().UnixNano())))
	141	+	wg.Go(func() {
	142	+	for sched.Err() == nil {
	143	+	send(job{req: pk.pick(rng)})
	144	+	}
	145	+	})
	146	+	}
	147	+	wg.Wait()
	148	+	return 0
	149	+	}
	150	+
	151	+	// The queue absorbs scheduling hiccups of the load generator; a request
	152	+	// that finds it full is missed.
	153	+	jobs := make(chan job, cfg.clients)
	154	+	for range cfg.clients {
	155	+	wg.Go(func() {
	156	+	for j := range jobs {
	157	+	send(j)
	158	+	}
	159	+	})
	160	+	}
	161	+	var missed int64
	162	+	rng := rand.New(rand.NewPCG(1, uint64(time.Now().UnixNano())))
	163	+	interval := time.Duration(float64(time.Second) / cfg.rate)
	164	+	next := time.Now()
	165	+	loop:
	166	+	for {
	167	+	if wait := time.Until(next); wait > 0 {
	168	+	select {
	169	+	case <-sched.Done():
	170	+	break loop
	171	+	case <-time.After(wait):
	172	+	}
	173	+	}
	174	+	if sched.Err() != nil {
	175	+	break
	176	+	}
	177	+	select {
	178	+	case jobs <- job{req: pk.pick(rng), at: next}:
	179	+	default:
	180	+	missed++
	181	+	}
	182	+	next = next.Add(interval)
	183	+	}
	184	+	close(jobs)
	185	+	wg.Wait()
	186	+	if collect == nil {
	187	+	return 0
	188	+	}
	189	+	return missed
	190	+	}
	191	+
	192	+	// do sends one request and reads the whole body.
	193	+	func do(ctx context.Context, client *http.Client, cfg config, base string, r Request) sample {
	194	+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, base+r.Path, nil)
	195	+	if err != nil {
	196	+	return sample{err: err.Error()}
	197	+	}
	198	+	header(req, cfg)
	199	+	t := time.Now()
	200	+	resp, err := client.Do(req)
	201	+	if err != nil {
	202	+	return sample{latency: time.Since(t), err: shortErr(err)}
	203	+	}
	204	+	n, err := io.Copy(io.Discard, resp.Body)
	205	+	_ = resp.Body.Close() // fully read: closing frees the connection
	206	+	s := sample{latency: time.Since(t), bytes: n}
	207	+	switch {
	208	+	case err != nil:
	209	+	s.err = shortErr(err)
	210	+	case resp.StatusCode != r.Status:
	211	+	s.err = fmt.Sprintf("status %d, want %d", resp.StatusCode, r.Status)
	212	+	}
	213	+	return s
	214	+	}
	215	+
	216	+	func header(req *http.Request, cfg config) {
	217	+	req.Header.Set("User-Agent", "ermite-loadtest")
	218	+	if cfg.gzip {
	219	+	req.Header.Set("Accept-Encoding", "gzip, br, zstd")
	220	+	}
	221	+	if cfg.token != "" {
	222	+	req.Header.Set("Authorization", "Bearer "+cfg.token)
	223	+	}
	224	+	}
	225	+
	226	+	func shortErr(err error) string {
	227	+	s := err.Error()
	228	+	if i := strings.LastIndex(s, ": "); i >= 0 {
	229	+	s = s[i+2:]
	230	+	}
	231	+	return s
	232	+	}
	233	+
	234	+	// streamConnectTimeout bounds the wait for the headers of a stream.
	235	+	const streamConnectTimeout = 5 * time.Second
	236	+
	237	+	// streamSet follows the open streams of a run.
	238	+	type streamSet struct {
	239	+	wg sync.WaitGroup
	240	+	stats []*streamStat
	241	+	}
	242	+
	243	+	type streamStat struct {
	244	+	name string
	245	+	want int
	246	+	opened atomic.Int64
	247	+	refused atomic.Int64 // a status other than 200, or a failed dial
	248	+	dropped atomic.Int64 // closed by the server before the end of the run
	249	+	events atomic.Int64
	250	+	ttfb struct {
	251	+	sync.Mutex
	252	+	all []time.Duration
	253	+	}
	254	+	}
	255	+
	256	+	func openStreams(ctx context.Context, cfg config, base string, sc Scenario) streamSet {
	257	+	set := &streamSet{}
	258	+	if cfg.streams <= 0 {
	259	+	return set
	260	+	}
	261	+	// Each stream is one visitor: its own connection and, against a loopback
	262	+	// server, its own source address (127.1.x.y), since the server bounds the
	263	+	// streams per client address.
	264	+	loopback := isLoopback(base)
	265	+	n := 0
	266	+	for _, s := range sc.Streams {
	267	+	st := &streamStat{name: s.Name, want: int(float64(s.Count)*cfg.streams + .5)}
	268	+	set.stats = append(set.stats, st)
	269	+	for range st.want {
	270	+	client := streamClient(loopback, n)
	271	+	n++
	272	+	set.wg.Go(func() { follow(ctx, client, cfg, base+s.Path, st) })
	273	+	}
	274	+	}
	275	+	// Let them connect before the measured run starts; a stream still not
	276	+	// open after streamConnectTimeout has failed (Report.Failed).
	277	+	deadline := time.Now().Add(streamConnectTimeout + time.Second)
	278	+	for time.Now().Before(deadline) {
	279	+	done := true
	280	+	for _, st := range set.stats {
	281	+	if int(st.opened.Load()+st.refused.Load()) < st.want {
	282	+	done = false
	283	+	}
	284	+	}
	285	+	if done {
	286	+	break
	287	+	}
	288	+	time.Sleep(20 * time.Millisecond)
	289	+	}
	290	+	return set
	291	+	}
	292	+
	293	+	func isLoopback(base string) bool {
	294	+	u, err := url.Parse(base)
	295	+	if err != nil {
	296	+	return false
	297	+	}
	298	+	ip, err := netip.ParseAddr(u.Hostname())
	299	+	return err == nil && ip.Is4() && ip.IsLoopback()
	300	+	}
	301	+
	302	+	// streamClient is the client of the n-th stream.
	303	+	func streamClient(loopback bool, n int) *http.Client {
	304	+	d := &net.Dialer{Timeout: 5 * time.Second}
	305	+	if loopback {
	306	+	src := netip.AddrFrom4([4]byte{127, 1, byte(n / 250), byte(n%250 + 1)})
	307	+	d.LocalAddr = &net.TCPAddr{IP: src.AsSlice()}
	308	+	}
	309	+	// A stream is a single request: no keep-alive, so that a stream the server
	310	+	// closes frees its connection at once. A stream whose headers do not come
	311	+	// in time counts as refused.
	312	+	return &http.Client{Transport: &http.Transport{
	313	+	DialContext: d.DialContext,
	314	+	DisableCompression: true,
	315	+	DisableKeepAlives: true,
	316	+	ResponseHeaderTimeout: streamConnectTimeout,
	317	+	}}
	318	+	}
	319	+
	320	+	func follow(ctx context.Context, client http.Client, cfg config, target string, st streamStat) {
	321	+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, target, nil)
	322	+	if err != nil {
	323	+	st.refused.Add(1)
	324	+	return
	325	+	}
	326	+	header(req, cfg)
	327	+	req.Header.Del("Accept-Encoding")
	328	+	req.Header.Set("Accept", "text/event-stream")
	329	+	t := time.Now()
	330	+	resp, err := client.Do(req)
	331	+	if err != nil {
	332	+	if ctx.Err() == nil {
	333	+	st.refused.Add(1)
	334	+	}
	335	+	return
	336	+	}
	337	+	defer func() { _ = resp.Body.Close() }()
	338	+	if resp.StatusCode != http.StatusOK {
	339	+	st.refused.Add(1)
	340	+	return
	341	+	}
	342	+	st.opened.Add(1)
	343	+	st.ttfb.Lock()
	344	+	st.ttfb.all = append(st.ttfb.all, time.Since(t))
	345	+	st.ttfb.Unlock()
	346	+	sc := bufio.NewScanner(resp.Body)
	347	+	sc.Buffer(make([]byte, 64<<10), 4<<20)
	348	+	for sc.Scan() {
	349	+	if strings.HasPrefix(sc.Text(), "event:") {
	350	+	st.events.Add(1)
	351	+	}
	352	+	}
	353	+	if ctx.Err() == nil {
	354	+	st.dropped.Add(1)
	355	+	}
	356	+	}
	357	+
	358	+	func (s *streamSet) wait() []StreamReport {
	359	+	s.wg.Wait()
	360	+	out := make([]StreamReport, 0, len(s.stats))
	361	+	for _, st := range s.stats {
	362	+	out = append(out, StreamReport{
	363	+	Name: st.name, Wanted: st.want,
	364	+	Opened: int(st.opened.Load()), Refused: int(st.refused.Load()),
	365	+	Dropped: int(st.dropped.Load()), Events: int(st.events.Load()),
	366	+	Connect: summarize(st.ttfb.all),
	367	+	})
	368	+	}
	369	+	return out
	370	+	}

@@ -0,0 +1,370 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package main
	4	+
	5	+	import (
	6	+	"bufio"
	7	+	"context"
	8	+	"errors"
	9	+	"fmt"
	10	+	"io"
	11	+	"math/rand/v2"
	12	+	"net"
	13	+	"net/http"
	14	+	"net/netip"
	15	+	"net/url"
	16	+	"strings"
	17	+	"sync"
	18	+	"sync/atomic"
	19	+	"time"
	20	+	)
	21	+
	22	+	type config struct {
	23	+	base string
	24	+	scenario string
	25	+	vars varsFlag
	26	+	clients int
	27	+	rate float64
	28	+	duration time.Duration
	29	+	warmup time.Duration
	30	+	streams float64
	31	+	token string
	32	+	gzip bool
	33	+	jsonOut string
	34	+	}
	35	+
	36	+	// sample is one finished request.
	37	+	type sample struct {
	38	+	req int
	39	+	latency time.Duration
	40	+	bytes int64
	41	+	err string // "" when the request got the expected status
	42	+	}
	43	+
	44	+	// job is one request to send; at is when it was due, so that open-loop
	45	+	// latencies include the time spent waiting for a client (no coordinated
	46	+	// omission).
	47	+	type job struct {
	48	+	req int
	49	+	at time.Time
	50	+	}
	51	+
	52	+	func run(ctx context.Context, cfg config, sc Scenario) (Report, error) {
	53	+	base := strings.TrimRight(cfg.base, "/")
	54	+	tr := &http.Transport{
	55	+	MaxIdleConns: cfg.clients * 2,
	56	+	MaxIdleConnsPerHost: cfg.clients * 2,
	57	+	DisableCompression: true, // the Accept-Encoding header is set by hand
	58	+	IdleConnTimeout: 90 * time.Second,
	59	+	}
	60	+	client := &http.Client{Transport: tr, Timeout: 30 * time.Second}
	61	+	defer tr.CloseIdleConnections()
	62	+
	63	+	if err := preflight(ctx, client, cfg, base, sc); err != nil {
	64	+	return nil, err
	65	+	}
	66	+
	67	+	// Streams are opened first and stay open for the whole run.
	68	+	sctx, scancel := context.WithCancel(ctx)
	69	+	defer scancel()
	70	+	streamStats := openStreams(sctx, cfg, base, sc)
	71	+
	72	+	if cfg.warmup > 0 {
	73	+	fire(ctx, client, cfg, base, sc, cfg.warmup, nil)
	74	+	}
	75	+	var mu sync.Mutex
	76	+	var samples []sample
	77	+	collect := func(s sample) {
	78	+	mu.Lock()
	79	+	samples = append(samples, s)
	80	+	mu.Unlock()
	81	+	}
	82	+	start := time.Now()
	83	+	missed := fire(ctx, client, cfg, base, sc, cfg.duration, collect)
	84	+	elapsed := time.Since(start)
	85	+	scancel()
	86	+	streams := streamStats.wait()
	87	+	if err := ctx.Err(); err != nil {
	88	+	return nil, fmt.Errorf("interrupted after %v: %w", elapsed.Round(time.Millisecond), err)
	89	+	}
	90	+
	91	+	return newReport(cfg, sc, samples, missed, elapsed, streams), nil
	92	+	}
	93	+
	94	+	// preflight fetches every request once and fails fast on a wrong status, so
	95	+	// that a typo in a scenario does not turn into a run of errors.
	96	+	func preflight(ctx context.Context, client http.Client, cfg config, base string, sc Scenario) error {
	97	+	var errs []string
	98	+	for _, r := range sc.Requests {
	99	+	s := do(ctx, client, cfg, base, r)
	100	+	if s.err != "" {
	101	+	errs = append(errs, fmt.Sprintf("%s (%s): %s", r.Name, r.Path, s.err))
	102	+	}
	103	+	}
	104	+	if len(errs) > 0 {
	105	+	return errors.New("preflight failed:\n " + strings.Join(errs, "\n "))
	106	+	}
	107	+	return nil
	108	+	}
	109	+
	110	+	// fire sends requests for d, closed-loop or open-loop, and hands each result
	111	+	// to collect (nil during the warm-up). It returns the open-loop requests that
	112	+	// found no free client.
	113	+	//
	114	+	// The deadline only stops new requests: those in flight or queued when it
	115	+	// passes still run to the end (bounded by the client timeout) and are
	116	+	// counted, so that slow responses are not dropped from the figures. Only an
	117	+	// interruption of ctx cancels them.
	118	+	func fire(ctx context.Context, client http.Client, cfg config, base string, sc Scenario, d time.Duration, collect func(sample)) int64 {
	119	+	sched, cancel := context.WithTimeout(ctx, d)
	120	+	defer cancel()
	121	+	pk := newPicker(sc.Requests)
	122	+	var wg sync.WaitGroup
	123	+
	124	+	send := func(j job) {
	125	+	s := do(ctx, client, cfg, base, sc.Requests[j.req])
	126	+	if ctx.Err() != nil {
	127	+	return // interrupted: the run reports no figures
	128	+	}
	129	+	s.req = j.req
	130	+	if !j.at.IsZero() {
	131	+	s.latency = time.Since(j.at)
	132	+	}
	133	+	if collect != nil {
	134	+	collect(s)
	135	+	}
	136	+	}
	137	+
	138	+	if cfg.rate <= 0 {
	139	+	for i := range cfg.clients {
	140	+	rng := rand.New(rand.NewPCG(uint64(i), uint64(time.Now().UnixNano())))
	141	+	wg.Go(func() {
	142	+	for sched.Err() == nil {
	143	+	send(job{req: pk.pick(rng)})
	144	+	}
	145	+	})
	146	+	}
	147	+	wg.Wait()
	148	+	return 0
	149	+	}
	150	+
	151	+	// The queue absorbs scheduling hiccups of the load generator; a request
	152	+	// that finds it full is missed.
	153	+	jobs := make(chan job, cfg.clients)
	154	+	for range cfg.clients {
	155	+	wg.Go(func() {
	156	+	for j := range jobs {
	157	+	send(j)
	158	+	}
	159	+	})
	160	+	}
	161	+	var missed int64
	162	+	rng := rand.New(rand.NewPCG(1, uint64(time.Now().UnixNano())))
	163	+	interval := time.Duration(float64(time.Second) / cfg.rate)
	164	+	next := time.Now()
	165	+	loop:
	166	+	for {
	167	+	if wait := time.Until(next); wait > 0 {
	168	+	select {
	169	+	case <-sched.Done():
	170	+	break loop
	171	+	case <-time.After(wait):
	172	+	}
	173	+	}
	174	+	if sched.Err() != nil {
	175	+	break
	176	+	}
	177	+	select {
	178	+	case jobs <- job{req: pk.pick(rng), at: next}:
	179	+	default:
	180	+	missed++
	181	+	}
	182	+	next = next.Add(interval)
	183	+	}
	184	+	close(jobs)
	185	+	wg.Wait()
	186	+	if collect == nil {
	187	+	return 0
	188	+	}
	189	+	return missed
	190	+	}
	191	+
	192	+	// do sends one request and reads the whole body.
	193	+	func do(ctx context.Context, client *http.Client, cfg config, base string, r Request) sample {
	194	+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, base+r.Path, nil)
	195	+	if err != nil {
	196	+	return sample{err: err.Error()}
	197	+	}
	198	+	header(req, cfg)
	199	+	t := time.Now()
	200	+	resp, err := client.Do(req)
	201	+	if err != nil {
	202	+	return sample{latency: time.Since(t), err: shortErr(err)}
	203	+	}
	204	+	n, err := io.Copy(io.Discard, resp.Body)
	205	+	_ = resp.Body.Close() // fully read: closing frees the connection
	206	+	s := sample{latency: time.Since(t), bytes: n}
	207	+	switch {
	208	+	case err != nil:
	209	+	s.err = shortErr(err)
	210	+	case resp.StatusCode != r.Status:
	211	+	s.err = fmt.Sprintf("status %d, want %d", resp.StatusCode, r.Status)
	212	+	}
	213	+	return s
	214	+	}
	215	+
	216	+	func header(req *http.Request, cfg config) {
	217	+	req.Header.Set("User-Agent", "ermite-loadtest")
	218	+	if cfg.gzip {
	219	+	req.Header.Set("Accept-Encoding", "gzip, br, zstd")
	220	+	}
	221	+	if cfg.token != "" {
	222	+	req.Header.Set("Authorization", "Bearer "+cfg.token)
	223	+	}
	224	+	}
	225	+
	226	+	func shortErr(err error) string {
	227	+	s := err.Error()
	228	+	if i := strings.LastIndex(s, ": "); i >= 0 {
	229	+	s = s[i+2:]
	230	+	}
	231	+	return s
	232	+	}
	233	+
	234	+	// streamConnectTimeout bounds the wait for the headers of a stream.
	235	+	const streamConnectTimeout = 5 * time.Second
	236	+
	237	+	// streamSet follows the open streams of a run.
	238	+	type streamSet struct {
	239	+	wg sync.WaitGroup
	240	+	stats []*streamStat
	241	+	}
	242	+
	243	+	type streamStat struct {
	244	+	name string
	245	+	want int
	246	+	opened atomic.Int64
	247	+	refused atomic.Int64 // a status other than 200, or a failed dial
	248	+	dropped atomic.Int64 // closed by the server before the end of the run
	249	+	events atomic.Int64
	250	+	ttfb struct {
	251	+	sync.Mutex
	252	+	all []time.Duration
	253	+	}
	254	+	}
	255	+
	256	+	func openStreams(ctx context.Context, cfg config, base string, sc Scenario) streamSet {
	257	+	set := &streamSet{}
	258	+	if cfg.streams <= 0 {
	259	+	return set
	260	+	}
	261	+	// Each stream is one visitor: its own connection and, against a loopback
	262	+	// server, its own source address (127.1.x.y), since the server bounds the
	263	+	// streams per client address.
	264	+	loopback := isLoopback(base)
	265	+	n := 0
	266	+	for _, s := range sc.Streams {
	267	+	st := &streamStat{name: s.Name, want: int(float64(s.Count)*cfg.streams + .5)}
	268	+	set.stats = append(set.stats, st)
	269	+	for range st.want {
	270	+	client := streamClient(loopback, n)
	271	+	n++
	272	+	set.wg.Go(func() { follow(ctx, client, cfg, base+s.Path, st) })
	273	+	}
	274	+	}
	275	+	// Let them connect before the measured run starts; a stream still not
	276	+	// open after streamConnectTimeout has failed (Report.Failed).
	277	+	deadline := time.Now().Add(streamConnectTimeout + time.Second)
	278	+	for time.Now().Before(deadline) {
	279	+	done := true
	280	+	for _, st := range set.stats {
	281	+	if int(st.opened.Load()+st.refused.Load()) < st.want {
	282	+	done = false
	283	+	}
	284	+	}
	285	+	if done {
	286	+	break
	287	+	}
	288	+	time.Sleep(20 * time.Millisecond)
	289	+	}
	290	+	return set
	291	+	}
	292	+
	293	+	func isLoopback(base string) bool {
	294	+	u, err := url.Parse(base)
	295	+	if err != nil {
	296	+	return false
	297	+	}
	298	+	ip, err := netip.ParseAddr(u.Hostname())
	299	+	return err == nil && ip.Is4() && ip.IsLoopback()
	300	+	}
	301	+
	302	+	// streamClient is the client of the n-th stream.
	303	+	func streamClient(loopback bool, n int) *http.Client {
	304	+	d := &net.Dialer{Timeout: 5 * time.Second}
	305	+	if loopback {
	306	+	src := netip.AddrFrom4([4]byte{127, 1, byte(n / 250), byte(n%250 + 1)})
	307	+	d.LocalAddr = &net.TCPAddr{IP: src.AsSlice()}
	308	+	}
	309	+	// A stream is a single request: no keep-alive, so that a stream the server
	310	+	// closes frees its connection at once. A stream whose headers do not come
	311	+	// in time counts as refused.
	312	+	return &http.Client{Transport: &http.Transport{
	313	+	DialContext: d.DialContext,
	314	+	DisableCompression: true,
	315	+	DisableKeepAlives: true,
	316	+	ResponseHeaderTimeout: streamConnectTimeout,
	317	+	}}
	318	+	}
	319	+
	320	+	func follow(ctx context.Context, client http.Client, cfg config, target string, st streamStat) {
	321	+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, target, nil)
	322	+	if err != nil {
	323	+	st.refused.Add(1)
	324	+	return
	325	+	}
	326	+	header(req, cfg)
	327	+	req.Header.Del("Accept-Encoding")
	328	+	req.Header.Set("Accept", "text/event-stream")
	329	+	t := time.Now()
	330	+	resp, err := client.Do(req)
	331	+	if err != nil {
	332	+	if ctx.Err() == nil {
	333	+	st.refused.Add(1)
	334	+	}
	335	+	return
	336	+	}
	337	+	defer func() { _ = resp.Body.Close() }()
	338	+	if resp.StatusCode != http.StatusOK {
	339	+	st.refused.Add(1)
	340	+	return
	341	+	}
	342	+	st.opened.Add(1)
	343	+	st.ttfb.Lock()
	344	+	st.ttfb.all = append(st.ttfb.all, time.Since(t))
	345	+	st.ttfb.Unlock()
	346	+	sc := bufio.NewScanner(resp.Body)
	347	+	sc.Buffer(make([]byte, 64<<10), 4<<20)
	348	+	for sc.Scan() {
	349	+	if strings.HasPrefix(sc.Text(), "event:") {
	350	+	st.events.Add(1)
	351	+	}
	352	+	}
	353	+	if ctx.Err() == nil {
	354	+	st.dropped.Add(1)
	355	+	}
	356	+	}
	357	+
	358	+	func (s *streamSet) wait() []StreamReport {
	359	+	s.wg.Wait()
	360	+	out := make([]StreamReport, 0, len(s.stats))
	361	+	for _, st := range s.stats {
	362	+	out = append(out, StreamReport{
	363	+	Name: st.name, Wanted: st.want,
	364	+	Opened: int(st.opened.Load()), Refused: int(st.refused.Load()),
	365	+	Dropped: int(st.dropped.Load()), Events: int(st.events.Load()),
	366	+	Connect: summarize(st.ttfb.all),
	367	+	})
	368	+	}
	369	+	return out
	370	+	}

A scripts/loadtest/scenario.go +144 -0

@@ -0,0 +1,144 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package main
	4	+
	5	+	import (
	6	+	"embed"
	7	+	"encoding/json"
	8	+	"errors"
	9	+	"fmt"
	10	+	"math/rand/v2"
	11	+	"os"
	12	+	"path"
	13	+	"regexp"
	14	+	"strings"
	15	+	)
	16	+
	17	+	// A Scenario is the traffic of one or more pages: weighted page requests and
	18	+	// Server-Sent Events streams kept open while the requests run, as the pages
	19	+	// of real visitors do.
	20	+	type Scenario struct {
	21	+	Name string `json:"name"`
	22	+	Description string `json:"description"`
	23	+	Requests []Request `json:"requests"`
	24	+	Streams []Stream `json:"streams"`
	25	+	}
	26	+
	27	+	// A Request is one page or asset fetched over and over. Weight is its share
	28	+	// of the requests of the scenario.
	29	+	type Request struct {
	30	+	Name string `json:"name"`
	31	+	Path string `json:"path"`
	32	+	Weight int `json:"weight"`
	33	+	// Status is the expected status code, 200 when absent.
	34	+	Status int `json:"status"`
	35	+	}
	36	+
	37	+	// A Stream is a live updates feed (GET /-/events) held open during the run.
	38	+	// Count is how many are opened, scaled by the -streams flag.
	39	+	type Stream struct {
	40	+	Name string `json:"name"`
	41	+	Path string `json:"path"`
	42	+	Count int `json:"count"`
	43	+	}
	44	+
	45	+	//go:embed scenarios/*.json
	46	+	var builtin embed.FS
	47	+
	48	+	var varRef = regexp.MustCompile(`\{\{([a-z_]+)\}\}`)
	49	+
	50	+	// loadScenario reads a built-in scenario by name, or a JSON file by path,
	51	+	// and replaces the {{name}} references of its paths with vars.
	52	+	func loadScenario(ref string, vars map[string]string) (*Scenario, error) {
	53	+	var data []byte
	54	+	var err error
	55	+	if strings.HasSuffix(ref, ".json") {
	56	+	data, err = os.ReadFile(ref)
	57	+	} else {
	58	+	data, err = builtin.ReadFile(path.Join("scenarios", ref+".json"))
	59	+	}
	60	+	if err != nil {
	61	+	return nil, fmt.Errorf("scenario %q: %w (built-in: %s)", ref, err, strings.Join(builtinNames(), ", "))
	62	+	}
	63	+	var sc Scenario
	64	+	if err := json.Unmarshal(data, &sc); err != nil {
	65	+	return nil, fmt.Errorf("scenario %q: %w", ref, err)
	66	+	}
	67	+	return &sc, sc.resolve(vars)
	68	+	}
	69	+
	70	+	func builtinNames() []string {
	71	+	entries, _ := builtin.ReadDir("scenarios")
	72	+	names := make([]string, 0, len(entries))
	73	+	for _, e := range entries {
	74	+	names = append(names, strings.TrimSuffix(e.Name(), ".json"))
	75	+	}
	76	+	return names
	77	+	}
	78	+
	79	+	// resolve checks the scenario and expands its variables.
	80	+	func (sc *Scenario) resolve(vars map[string]string) error {
	81	+	if len(sc.Requests) == 0 {
	82	+	return errors.New("scenario has no requests")
	83	+	}
	84	+	var missing []string
	85	+	expand := func(p string) string {
	86	+	return varRef.ReplaceAllStringFunc(p, func(m string) string {
	87	+	name := varRef.FindStringSubmatch(m)[1]
	88	+	v, ok := vars[name]
	89	+	if !ok {
	90	+	missing = append(missing, name)
	91	+	}
	92	+	return v
	93	+	})
	94	+	}
	95	+	for i := range sc.Requests {
	96	+	r := &sc.Requests[i]
	97	+	if r.Weight <= 0 {
	98	+	r.Weight = 1
	99	+	}
	100	+	if r.Status == 0 {
	101	+	r.Status = 200
	102	+	}
	103	+	if r.Name == "" {
	104	+	r.Name = r.Path
	105	+	}
	106	+	r.Path = expand(r.Path)
	107	+	}
	108	+	for i := range sc.Streams {
	109	+	s := &sc.Streams[i]
	110	+	if s.Name == "" {
	111	+	s.Name = s.Path
	112	+	}
	113	+	s.Path = expand(s.Path)
	114	+	}
	115	+	if len(missing) > 0 {
	116	+	return fmt.Errorf("missing -var for %s", strings.Join(missing, ", "))
	117	+	}
	118	+	return nil
	119	+	}
	120	+
	121	+	// picker draws requests in proportion to their weights.
	122	+	type picker struct {
	123	+	reqs []Request
	124	+	total int
	125	+	}
	126	+
	127	+	func newPicker(reqs []Request) picker {
	128	+	p := picker{reqs: reqs}
	129	+	for _, r := range reqs {
	130	+	p.total += r.Weight
	131	+	}
	132	+	return p
	133	+	}
	134	+
	135	+	func (p picker) pick(rng *rand.Rand) int {
	136	+	n := rng.IntN(p.total)
	137	+	for i, r := range p.reqs {
	138	+	if n < r.Weight {
	139	+	return i
	140	+	}
	141	+	n -= r.Weight
	142	+	}
	143	+	return len(p.reqs) - 1
	144	+	}

@@ -0,0 +1,144 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package main
	4	+
	5	+	import (
	6	+	"embed"
	7	+	"encoding/json"
	8	+	"errors"
	9	+	"fmt"
	10	+	"math/rand/v2"
	11	+	"os"
	12	+	"path"
	13	+	"regexp"
	14	+	"strings"
	15	+	)
	16	+
	17	+	// A Scenario is the traffic of one or more pages: weighted page requests and
	18	+	// Server-Sent Events streams kept open while the requests run, as the pages
	19	+	// of real visitors do.
	20	+	type Scenario struct {
	21	+	Name string `json:"name"`
	22	+	Description string `json:"description"`
	23	+	Requests []Request `json:"requests"`
	24	+	Streams []Stream `json:"streams"`
	25	+	}
	26	+
	27	+	// A Request is one page or asset fetched over and over. Weight is its share
	28	+	// of the requests of the scenario.
	29	+	type Request struct {
	30	+	Name string `json:"name"`
	31	+	Path string `json:"path"`
	32	+	Weight int `json:"weight"`
	33	+	// Status is the expected status code, 200 when absent.
	34	+	Status int `json:"status"`
	35	+	}
	36	+
	37	+	// A Stream is a live updates feed (GET /-/events) held open during the run.
	38	+	// Count is how many are opened, scaled by the -streams flag.
	39	+	type Stream struct {
	40	+	Name string `json:"name"`
	41	+	Path string `json:"path"`
	42	+	Count int `json:"count"`
	43	+	}
	44	+
	45	+	//go:embed scenarios/*.json
	46	+	var builtin embed.FS
	47	+
	48	+	var varRef = regexp.MustCompile(`\{\{([a-z_]+)\}\}`)
	49	+
	50	+	// loadScenario reads a built-in scenario by name, or a JSON file by path,
	51	+	// and replaces the {{name}} references of its paths with vars.
	52	+	func loadScenario(ref string, vars map[string]string) (*Scenario, error) {
	53	+	var data []byte
	54	+	var err error
	55	+	if strings.HasSuffix(ref, ".json") {
	56	+	data, err = os.ReadFile(ref)
	57	+	} else {
	58	+	data, err = builtin.ReadFile(path.Join("scenarios", ref+".json"))
	59	+	}
	60	+	if err != nil {
	61	+	return nil, fmt.Errorf("scenario %q: %w (built-in: %s)", ref, err, strings.Join(builtinNames(), ", "))
	62	+	}
	63	+	var sc Scenario
	64	+	if err := json.Unmarshal(data, &sc); err != nil {
	65	+	return nil, fmt.Errorf("scenario %q: %w", ref, err)
	66	+	}
	67	+	return &sc, sc.resolve(vars)
	68	+	}
	69	+
	70	+	func builtinNames() []string {
	71	+	entries, _ := builtin.ReadDir("scenarios")
	72	+	names := make([]string, 0, len(entries))
	73	+	for _, e := range entries {
	74	+	names = append(names, strings.TrimSuffix(e.Name(), ".json"))
	75	+	}
	76	+	return names
	77	+	}
	78	+
	79	+	// resolve checks the scenario and expands its variables.
	80	+	func (sc *Scenario) resolve(vars map[string]string) error {
	81	+	if len(sc.Requests) == 0 {
	82	+	return errors.New("scenario has no requests")
	83	+	}
	84	+	var missing []string
	85	+	expand := func(p string) string {
	86	+	return varRef.ReplaceAllStringFunc(p, func(m string) string {
	87	+	name := varRef.FindStringSubmatch(m)[1]
	88	+	v, ok := vars[name]
	89	+	if !ok {
	90	+	missing = append(missing, name)
	91	+	}
	92	+	return v
	93	+	})
	94	+	}
	95	+	for i := range sc.Requests {
	96	+	r := &sc.Requests[i]
	97	+	if r.Weight <= 0 {
	98	+	r.Weight = 1
	99	+	}
	100	+	if r.Status == 0 {
	101	+	r.Status = 200
	102	+	}
	103	+	if r.Name == "" {
	104	+	r.Name = r.Path
	105	+	}
	106	+	r.Path = expand(r.Path)
	107	+	}
	108	+	for i := range sc.Streams {
	109	+	s := &sc.Streams[i]
	110	+	if s.Name == "" {
	111	+	s.Name = s.Path
	112	+	}
	113	+	s.Path = expand(s.Path)
	114	+	}
	115	+	if len(missing) > 0 {
	116	+	return fmt.Errorf("missing -var for %s", strings.Join(missing, ", "))
	117	+	}
	118	+	return nil
	119	+	}
	120	+
	121	+	// picker draws requests in proportion to their weights.
	122	+	type picker struct {
	123	+	reqs []Request
	124	+	total int
	125	+	}
	126	+
	127	+	func newPicker(reqs []Request) picker {
	128	+	p := picker{reqs: reqs}
	129	+	for _, r := range reqs {
	130	+	p.total += r.Weight
	131	+	}
	132	+	return p
	133	+	}
	134	+
	135	+	func (p picker) pick(rng *rand.Rand) int {
	136	+	n := rng.IntN(p.total)
	137	+	for i, r := range p.reqs {
	138	+	if n < r.Weight {
	139	+	return i
	140	+	}
	141	+	n -= r.Weight
	142	+	}
	143	+	return len(p.reqs) - 1
	144	+	}

A scripts/loadtest/scenarios/home.json +10 -0

@@ -0,0 +1,10 @@
	1	+	{
	2	+	"name": "home",
	3	+	"description": "Anonymous visitors on the home page, each keeping its live updates stream open.",
	4	+	"requests": [
	5	+	{ "name": "home", "path": "/", "weight": 1 }
	6	+	],
	7	+	"streams": [
	8	+	{ "name": "events home", "path": "/-/events?topic=home", "count": 200 }
	9	+	]
	10	+	}

@@ -0,0 +1,10 @@
	1	+	{
	2	+	"name": "home",
	3	+	"description": "Anonymous visitors on the home page, each keeping its live updates stream open.",
	4	+	"requests": [
	5	+	{ "name": "home", "path": "/", "weight": 1 }
	6	+	],
	7	+	"streams": [
	8	+	{ "name": "events home", "path": "/-/events?topic=home", "count": 200 }
	9	+	]
	10	+	}

A scripts/loadtest/serve.sh +88 -0

@@ -0,0 +1,88 @@
	1	+	#!/usr/bin/env bash
	2	+	# SPDX-License-Identifier: AGPL-3.0-or-later
	3	+	#
	4	+	# Builds ermite from this checkout and starts a throw-away server for the load
	5	+	# tests, seeded with repositories whose history spreads over 90 days. The
	6	+	# server is pinned to CPUS cores (GOMAXPROCS and taskset), so that the figures
	7	+	# say what a small host can do, not what the workstation can.
	8	+	#
	9	+	# scripts/loadtest/serve.sh # 1 core, http://127.0.0.1:7418
	10	+	# CPUS=2 scripts/loadtest/serve.sh
	11	+	# scripts/loadtest/serve.sh stop
	12	+	set -euo pipefail
	13	+
	14	+	ROOT=${LOADTEST_DIR:-${TMPDIR:-/tmp}/ermite-loadtest}
	15	+	ADDR=${ADDR:-127.0.0.1:7418}
	16	+	CPUS=${CPUS:-1}
	17	+	URL=http://$ADDR
	18	+	TOKEN=loadtest-admin-token-0123456789abcdefghijklmnop
	19	+
	20	+	# stop ends the server and waits for it (it shuts down gracefully), so that
	21	+	# its data is never removed under it.
	22	+	stop() {
	23	+	[[ -f $ROOT/pid ]] \|\| return 0
	24	+	local pid; pid=$(cat "$ROOT/pid")
	25	+	if kill "$pid" 2>/dev/null; then
	26	+	for _ in $(seq 150); do kill -0 "$pid" 2>/dev/null \|\| break; sleep 0.1; done
	27	+	kill -9 "$pid" 2>/dev/null \|\| true
	28	+	fi
	29	+	rm -f "$ROOT/pid"
	30	+	}
	31	+	if [[ ${1:-} == stop ]]; then stop; exit 0; fi
	32	+	stop
	33	+
	34	+	cd "$(git rev-parse --show-toplevel)"
	35	+	rm -rf "$ROOT" && mkdir -p "$ROOT/data" "$ROOT/wt"
	36	+	GOWORK=off go build -o "$ROOT/ermite" ./cmd/ermite
	37	+	BIN=$ROOT/ermite
	38	+
	39	+	export ERMITE_DATA_DIR=$ROOT/data ERMITE_PUBLIC_URL=$URL ERMITE_ADDR=$ADDR
	40	+	export ERMITE_ADMIN_TOKEN=$TOKEN ERMITE_LOG_FORMAT=console ERMITE_LOG_LEVEL=warn
	41	+	export ERMITE_URL=$URL ERMITE_TOKEN=$TOKEN ERMITE_CONFIG=$ROOT/cfg
	42	+
	43	+	# The last CPUS of the CPUs this shell may use run the server; run the load
	44	+	# generator on the others (see README.md).
	45	+	mapfile -t allowed < <(python3 -c 'import os; print(*sorted(os.sched_getaffinity(0)), sep="\n")')
	46	+	if ! [[ $CPUS =~ ^[1-9][0-9]*$ ]] \|\| ((CPUS >= ${#allowed[@]})); then
	47	+	echo "CPUS must be between 1 and $((${#allowed[@]} - 1)), got $CPUS" >&2; exit 2
	48	+	fi
	49	+	server_cpus=$(IFS=,; echo "${allowed[*]: -CPUS}")
	50	+	GOMAXPROCS=$CPUS taskset -c "$server_cpus" "$BIN" serve >"$ROOT/log" 2>&1 &
	51	+	echo $! >"$ROOT/pid"
	52	+	for _ in $(seq 100); do curl -fs "$URL/-/health" >/dev/null && break; sleep 0.1; done
	53	+
	54	+	now=$(date +%s)
	55	+	seed() { # name visibility description profile
	56	+	local name=$1 vis=$2 desc=$3 prof=$4 d=$ROOT/wt/$1
	57	+	"$BIN" repo create "$name" --visibility "$vis" --description "$desc" >/dev/null
	58	+	mkdir -p "$d" && cd "$d"
	59	+	git init -q -b main . && git config user.email loadtest@example.test && git config user.name Loadtest
	60	+	python3 - "$prof" "$now" >"$ROOT/$name.dates" <<'PY'
	61	+	import sys, math, random
	62	+	prof, now = sys.argv[1], int(sys.argv[2]); random.seed(prof)
	63	+	out = []
	64	+	for day in range(89, -1, -1):
	65	+	if prof == 'busy': lam = 6 + 5*math.sin(day/9) + (14 if day < 6 else 0)
	66	+	elif prof == 'bursty': lam = 18 if 40 < day < 48 or 10 < day < 14 else 0.3
	67	+	elif prof == 'sleepy': lam = 3 if day > 60 else 0
	68	+	else: lam = 1.2
	69	+	for _ in range(max(0, int(random.gauss(lam, math.sqrt(lam + 0.1))))):
	70	+	out.append(now - day*86400 - random.randint(0, 80000))
	71	+	for t in sorted(out): print(t)
	72	+	PY
	73	+	local i=0 subjects=("initial import" "add the parser" "fix an off-by-one error in the tokenizer" "refactor storage layer" "bump dependencies" "document the public API" "tidy up")
	74	+	while read -r t; do
	75	+	i=$((i + 1)); echo $i >f.txt; git add -A
	76	+	GIT_AUTHOR_DATE="@$t" GIT_COMMITTER_DATE="@$t" git commit -qm "${subjects[$((i % 7))]} $i"
	77	+	done <"$ROOT/$name.dates"
	78	+	git tag "v1.$((i / 10)).0"
	79	+	git push -q --tags "http://x:$TOKEN@$ADDR/$name.git" main
	80	+	}
	81	+	seed alpha public "A small Wayland compositor that spends its frames on your apps" busy
	82	+	seed bravo public "Keyboard-driven web browser for tiling window managers" bursty
	83	+	seed charlie public "Self-hosted container deployment: push an image, deploy it" steady
	84	+	seed delta private "Library that reads terminal output and tracks the screen" sleepy
	85	+	"$BIN" repo create echo --visibility public --description "Never pushed" >/dev/null
	86	+
	87	+	echo "ermite on $URL, CPU(s) $server_cpus, data in $ROOT (pid $(cat "$ROOT/pid"))"
	88	+	echo "admin token: $TOKEN"

@@ -0,0 +1,88 @@
	1	+	#!/usr/bin/env bash
	2	+	# SPDX-License-Identifier: AGPL-3.0-or-later
	3	+	#
	4	+	# Builds ermite from this checkout and starts a throw-away server for the load
	5	+	# tests, seeded with repositories whose history spreads over 90 days. The
	6	+	# server is pinned to CPUS cores (GOMAXPROCS and taskset), so that the figures
	7	+	# say what a small host can do, not what the workstation can.
	8	+	#
	9	+	# scripts/loadtest/serve.sh # 1 core, http://127.0.0.1:7418
	10	+	# CPUS=2 scripts/loadtest/serve.sh
	11	+	# scripts/loadtest/serve.sh stop
	12	+	set -euo pipefail
	13	+
	14	+	ROOT=${LOADTEST_DIR:-${TMPDIR:-/tmp}/ermite-loadtest}
	15	+	ADDR=${ADDR:-127.0.0.1:7418}
	16	+	CPUS=${CPUS:-1}
	17	+	URL=http://$ADDR
	18	+	TOKEN=loadtest-admin-token-0123456789abcdefghijklmnop
	19	+
	20	+	# stop ends the server and waits for it (it shuts down gracefully), so that
	21	+	# its data is never removed under it.
	22	+	stop() {
	23	+	[[ -f $ROOT/pid ]] \|\| return 0
	24	+	local pid; pid=$(cat "$ROOT/pid")
	25	+	if kill "$pid" 2>/dev/null; then
	26	+	for _ in $(seq 150); do kill -0 "$pid" 2>/dev/null \|\| break; sleep 0.1; done
	27	+	kill -9 "$pid" 2>/dev/null \|\| true
	28	+	fi
	29	+	rm -f "$ROOT/pid"
	30	+	}
	31	+	if [[ ${1:-} == stop ]]; then stop; exit 0; fi
	32	+	stop
	33	+
	34	+	cd "$(git rev-parse --show-toplevel)"
	35	+	rm -rf "$ROOT" && mkdir -p "$ROOT/data" "$ROOT/wt"
	36	+	GOWORK=off go build -o "$ROOT/ermite" ./cmd/ermite
	37	+	BIN=$ROOT/ermite
	38	+
	39	+	export ERMITE_DATA_DIR=$ROOT/data ERMITE_PUBLIC_URL=$URL ERMITE_ADDR=$ADDR
	40	+	export ERMITE_ADMIN_TOKEN=$TOKEN ERMITE_LOG_FORMAT=console ERMITE_LOG_LEVEL=warn
	41	+	export ERMITE_URL=$URL ERMITE_TOKEN=$TOKEN ERMITE_CONFIG=$ROOT/cfg
	42	+
	43	+	# The last CPUS of the CPUs this shell may use run the server; run the load
	44	+	# generator on the others (see README.md).
	45	+	mapfile -t allowed < <(python3 -c 'import os; print(*sorted(os.sched_getaffinity(0)), sep="\n")')
	46	+	if ! [[ $CPUS =~ ^[1-9][0-9]*$ ]] \|\| ((CPUS >= ${#allowed[@]})); then
	47	+	echo "CPUS must be between 1 and $((${#allowed[@]} - 1)), got $CPUS" >&2; exit 2
	48	+	fi
	49	+	server_cpus=$(IFS=,; echo "${allowed[*]: -CPUS}")
	50	+	GOMAXPROCS=$CPUS taskset -c "$server_cpus" "$BIN" serve >"$ROOT/log" 2>&1 &
	51	+	echo $! >"$ROOT/pid"
	52	+	for _ in $(seq 100); do curl -fs "$URL/-/health" >/dev/null && break; sleep 0.1; done
	53	+
	54	+	now=$(date +%s)
	55	+	seed() { # name visibility description profile
	56	+	local name=$1 vis=$2 desc=$3 prof=$4 d=$ROOT/wt/$1
	57	+	"$BIN" repo create "$name" --visibility "$vis" --description "$desc" >/dev/null
	58	+	mkdir -p "$d" && cd "$d"
	59	+	git init -q -b main . && git config user.email loadtest@example.test && git config user.name Loadtest
	60	+	python3 - "$prof" "$now" >"$ROOT/$name.dates" <<'PY'
	61	+	import sys, math, random
	62	+	prof, now = sys.argv[1], int(sys.argv[2]); random.seed(prof)
	63	+	out = []
	64	+	for day in range(89, -1, -1):
	65	+	if prof == 'busy': lam = 6 + 5*math.sin(day/9) + (14 if day < 6 else 0)
	66	+	elif prof == 'bursty': lam = 18 if 40 < day < 48 or 10 < day < 14 else 0.3
	67	+	elif prof == 'sleepy': lam = 3 if day > 60 else 0
	68	+	else: lam = 1.2
	69	+	for _ in range(max(0, int(random.gauss(lam, math.sqrt(lam + 0.1))))):
	70	+	out.append(now - day*86400 - random.randint(0, 80000))
	71	+	for t in sorted(out): print(t)
	72	+	PY
	73	+	local i=0 subjects=("initial import" "add the parser" "fix an off-by-one error in the tokenizer" "refactor storage layer" "bump dependencies" "document the public API" "tidy up")
	74	+	while read -r t; do
	75	+	i=$((i + 1)); echo $i >f.txt; git add -A
	76	+	GIT_AUTHOR_DATE="@$t" GIT_COMMITTER_DATE="@$t" git commit -qm "${subjects[$((i % 7))]} $i"
	77	+	done <"$ROOT/$name.dates"
	78	+	git tag "v1.$((i / 10)).0"
	79	+	git push -q --tags "http://x:$TOKEN@$ADDR/$name.git" main
	80	+	}
	81	+	seed alpha public "A small Wayland compositor that spends its frames on your apps" busy
	82	+	seed bravo public "Keyboard-driven web browser for tiling window managers" bursty
	83	+	seed charlie public "Self-hosted container deployment: push an image, deploy it" steady
	84	+	seed delta private "Library that reads terminal output and tracks the screen" sleepy
	85	+	"$BIN" repo create echo --visibility public --description "Never pushed" >/dev/null
	86	+
	87	+	echo "ermite on $URL, CPU(s) $server_cpus, data in $ROOT (pid $(cat "$ROOT/pid"))"
	88	+	echo "admin token: $TOKEN"