[local_dss_bench] Add rid isa with one subscription

.gitignore

@@ -131,3 +131,5 @@ go
 
 # setuptools_scm
 _version.py
+
+dss_bench_out

monitoring/local_dss_bench/README.md

@@ -0,0 +1,44 @@
+# local_dss_bench
+
+Benchmark performance of a local DSS deployment as a function of deployment parameters.
+
+For every **context** in every sweep for each **test**, it: cleans the local ecosystem,
+`make start-locally` with the right env, runs the test against the DSS load balancer, and records q/s + latency percentiles. Output is one plot per
+test, q/s and latency vs context.
+
+## Layout
+
+- `config.py` - global settings (node count, image, datastore type, duration, processes).
+- `environment.py` - wraps `make start-locally` / `down-locally`.
+- `driver.py` - runs `processes` processes against the DSS load balancer for `duration`, times each call.
+- `measure.py` - aggregates into total q/s + median/p95 latency.
+- `plot.py` - one PNG per test (q/s + latency, one line per comparison arm).
+- `arms.py` - optional comparison arms (image vs image, or datastore vs datastore).
+- `run.py` - CLI matrix runner.
+- `sweeps/` - one file per sweep.
+  from 0 to 100 ms).
+- `tests/` - one file per test.
+
+## Run
+
+```bash
+uv run python -m monitoring.local_dss_bench.run --processes 8 --duration 30
+```
+
+## Compare (optional)
+
+Two arms, overlaid on every plot. Omit both flags for no comparison.
+
+```bash
+# two PRs / images
+uv run python -m monitoring.local_dss_bench.run --compare-images interuss/dss:pr-A interuss/dss:pr-B
+# two datastores
+uv run python -m monitoring.local_dss_bench.run --compare-datastores crdb raft
+```
+
+The bench runs on the host and reaches the DSS load balancer at
+`http://localhost:8090`.
+
+## Add a sweep or test
+
+Drop a new file in `sweeps/` (subclass `Sweep`) or `tests/` (subclass `BenchTest`).

monitoring/local_dss_bench/arms.py

@@ -0,0 +1,34 @@
+"""Optional comparison arms: run the whole matrix under several configs and
+overlay them on the plots. Default is a single arm (no comparison)."""
+
+import dataclasses
+from dataclasses import dataclass, field
+
+from monitoring.local_dss_bench.config import GlobalConfig
+
+
+@dataclass
+class Arm:
+    label: str
+    overrides: dict = field(default_factory=dict)
+
+    def apply(self, cfg: GlobalConfig) -> GlobalConfig:
+        return dataclasses.replace(cfg, **self.overrides)
+
+
+def single(cfg: GlobalConfig) -> list[Arm]:
+    return [Arm(label="baseline")]
+
+
+def compare_images(img_a: str, img_b: str) -> list[Arm]:
+    return [
+        Arm(label=img_a, overrides={"dss_image": img_a}),
+        Arm(label=img_b, overrides={"dss_image": img_b}),
+    ]
+
+
+def compare_datastores(db_a: str, db_b: str) -> list[Arm]:
+    return [
+        Arm(label=db_a, overrides={"db_type": db_a}),
+        Arm(label=db_b, overrides={"db_type": db_b}),
+    ]

monitoring/local_dss_bench/config.py

@@ -0,0 +1,22 @@
+"""Global, parameterizable settings for a benchmark run."""
+
+from dataclasses import dataclass
+
+
+@dataclass
+class GlobalConfig:
+    # Local ecosystem sizing (consumed by `make start-locally`).
+    num_uss: int = 3
+    num_nodes: int = 3
+    dss_image: str = "interuss/dss:v0.22.0"
+    db_type: str = "crdb"  # crdb | ybdb | raft
+    intra_netem: str = "delay 600us 40us 25% distribution normal loss 0.0005%"
+    inter_netem: str = "delay 36ms 40ms 50% distribution paretonormal loss 0.25% 15%"
+
+    # Load profile.
+    duration_s: float = 120.0
+    processes: int = 9  # parallel processes calling action()
+
+    # Dummy OAuth reachable from the host.
+    oauth_token_endpoint: str = "http://localhost:8085/token"
+    oauth_sub: str = "uss1"

monitoring/local_dss_bench/driver.py

@@ -0,0 +1,86 @@
+"""Run a BenchTest with `cfg.processes` processes hitting the DSS load
+balancer in parallel for `cfg.duration_s`. Each process is one sequential
+caller; q/s emerges from the number of processes."""
+
+import time
+from multiprocessing import Process, Queue
+
+from monitoring.local_dss_bench.config import GlobalConfig
+from monitoring.local_dss_bench.tests.base import BenchTest
+from monitoring.monitorlib.auth import DummyOAuth
+from monitoring.monitorlib.infrastructure import UTMClientSession
+
+
+def _build_session(
+    test: BenchTest,
+    base_url: str,
+    cfg: GlobalConfig,
+) -> UTMClientSession:
+    session = UTMClientSession(
+        base_url, DummyOAuth(cfg.oauth_token_endpoint, cfg.oauth_sub)
+    )
+    session.default_scopes = test.scopes
+    return session
+
+
+def _worker(
+    test: BenchTest,
+    base_url: str,
+    cfg: GlobalConfig,
+    q: Queue,
+) -> None:
+
+    session = _build_session(test, base_url, cfg)
+
+    try:
+        test.setup(session, base_url)
+    except Exception:
+        q.put((base_url, [], []))
+        return
+
+    latencies_ms: list[float] = []
+    error_latencies_ms: list[float] = []
+    end = time.monotonic() + cfg.duration_s
+    while time.monotonic() < end:
+        t0 = time.monotonic()
+        try:
+            test.action(session, base_url)
+            latencies_ms.append((time.monotonic() - t0) * 1000.0)
+        except Exception:
+            # Keep how long the failed call took (e.g. a ~10s timeout) instead
+            # of dropping it: discarding slow failures biases percentiles down.
+            error_latencies_ms.append((time.monotonic() - t0) * 1000.0)
+
+    try:
+        test.teardown(session, base_url)
+    except Exception:
+        pass
+
+    q.put((base_url, latencies_ms, error_latencies_ms))
+
+
+LB_URL = "http://localhost:8090"
+
+
+def run_test(test: BenchTest, cfg: GlobalConfig) -> dict[str, dict]:
+    """Return {base_url: {"latencies": [...ms], "error_latencies": [...ms]}}."""
+
+    session = _build_session(test, LB_URL, cfg)
+    test.prepare(session, LB_URL)
+
+    q: Queue = Queue()
+    procs = []
+    for _ in range(cfg.processes):
+        p = Process(target=_worker, args=(test, LB_URL, cfg, q))
+        p.start()
+        procs.append(p)
+
+    results: dict[str, dict] = {LB_URL: {"latencies": [], "error_latencies": []}}
+    for _ in procs:
+        url, lat, err_lat = q.get()
+        results[url]["latencies"].extend(lat)
+        results[url]["error_latencies"].extend(err_lat)
+    for p in procs:
+        p.join()
+
+    return results

monitoring/local_dss_bench/environment.py

@@ -0,0 +1,37 @@
+"""Drive `make start-locally` / `make down-locally`."""
+
+import os
+import subprocess
+from pathlib import Path
+
+from monitoring.local_dss_bench.config import GlobalConfig
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+
+
+def _env(cfg: GlobalConfig, extra: dict[str, str]) -> dict[str, str]:
+    env = dict(os.environ)
+    env.update(
+        {
+            "NUM_USS": str(cfg.num_uss),
+            "NUM_NODES": str(cfg.num_nodes),
+            "DSS_IMAGE": cfg.dss_image,
+            "DB_TYPE": cfg.db_type,
+            "INTRA_USS_NETEM_CONF": cfg.intra_netem,
+            "INTER_USS_NETEM_CONF": cfg.inter_netem,
+        }
+    )
+    env.update(extra)
+    return env
+
+
+def up(cfg: GlobalConfig, extra: dict[str, str]) -> None:
+    subprocess.run(
+        ["make", "start-locally"], cwd=REPO_ROOT, env=_env(cfg, extra), check=True
+    )
+
+
+def down(cfg: GlobalConfig) -> None:
+    subprocess.run(
+        ["make", "clean-locally"], cwd=REPO_ROOT, env=_env(cfg, {}), check=False
+    )

monitoring/local_dss_bench/measure.py

@@ -0,0 +1,58 @@
+"""Turn raw driver output into q/s and latency percentiles."""
+
+from dataclasses import dataclass
+
+
+def _percentile(values: list[float], pct: float) -> float:
+    if not values:
+        return 0.0
+    s = sorted(values)
+    k = (len(s) - 1) * (pct / 100.0)
+    lo = int(k)
+    hi = min(lo + 1, len(s) - 1)
+    return s[lo] + (s[hi] - s[lo]) * (k - lo)
+
+
+@dataclass
+class Datapoint:
+    label: str
+    rps_total: float
+    p50_ms: float
+    p95_ms: float
+    errors: int
+    err_p50_ms: float
+    err_p95_ms: float
+    p50_all_ms: float
+    p95_all_ms: float
+    rps_per_target: dict[str, float]
+
+
+def summarize(label: str, results: dict[str, dict], duration_s: float) -> Datapoint:
+    """Aggregate the pooled samples into total q/s plus median/p95 latency,
+    and keep per-target q/s."""
+    merged: list[float] = []
+    merged_errors: list[float] = []
+    per_target = {}
+    for url, d in results.items():
+        lat = d["latencies"]
+        merged.extend(lat)
+        merged_errors.extend(d["error_latencies"])
+        per_target[url] = round(len(lat) / duration_s, 2)
+
+    # Distribution including failed calls: a request that ran ~10s then timed
+    # out is a 10s+ latency, so folding error timings back in removes the
+    # survivorship bias of percentiles computed over successes only.
+    with_errors = merged + merged_errors
+
+    return Datapoint(
+        label=label,
+        rps_total=round(len(merged) / duration_s, 2),
+        p50_ms=round(_percentile(merged, 50), 2),
+        p95_ms=round(_percentile(merged, 95), 2),
+        errors=len(merged_errors),
+        err_p50_ms=round(_percentile(merged_errors, 50), 2),
+        err_p95_ms=round(_percentile(merged_errors, 95), 2),
+        p50_all_ms=round(_percentile(with_errors, 50), 2),
+        p95_all_ms=round(_percentile(with_errors, 95), 2),
+        rps_per_target=per_target,
+    )

monitoring/local_dss_bench/plot.py

@@ -0,0 +1,89 @@
+"""One plot per test, comparing arms. Top subplot: total q/s vs context.
+Bottom subplot: latency vs context (p50 dashed, p95 solid). One color per arm."""
+
+from pathlib import Path
+
+import matplotlib
+
+from monitoring.local_dss_bench.measure import Datapoint
+
+matplotlib.use("Agg")
+import matplotlib.pyplot as plt
+
+
+def render(
+    by_test: dict[str, dict[str, list[Datapoint]]],
+    outdir: Path,
+    axis_label: str = "context",
+    meta: dict | None = None,
+) -> list[Path]:
+    outdir.mkdir(parents=True, exist_ok=True)
+    meta_text = "  |  ".join(f"{k}={v}" for k, v in meta.items()) if meta else ""
+    paths = []
+    for test_name, by_arm in by_test.items():
+        fig, (ax_rps, ax_lat, ax_err) = plt.subplots(
+            3, 1, figsize=(10, 12), sharex=True
+        )
+        colors = plt.rcParams["axes.prop_cycle"].by_key()["color"]
+        for idx, (arm_label, points) in enumerate(by_arm.items()):
+            labels = [p.label for p in points]
+            color = colors[idx % len(colors)]
+            ax_rps.plot(
+                labels,
+                [p.rps_total for p in points],
+                marker="s",
+                color=color,
+                label=arm_label,
+            )
+            ax_lat.plot(
+                labels,
+                [p.p95_ms for p in points],
+                marker="o",
+                color=color,
+                label=f"{arm_label} p95",
+            )
+            ax_lat.plot(
+                labels,
+                [p.p50_ms for p in points],
+                marker="o",
+                linestyle="--",
+                color=color,
+                alpha=0.6,
+                label=f"{arm_label} p50",
+            )
+            ax_lat.plot(
+                labels,
+                [p.p95_all_ms for p in points],
+                marker=".",
+                linestyle=":",
+                color=color,
+                alpha=0.45,
+                label=f"{arm_label} p95 (incl. errors)",
+            )
+            ax_err.plot(
+                labels,
+                [p.errors for p in points],
+                marker="x",
+                color=color,
+                label=arm_label,
+            )
+
+        ax_rps.set_ylabel("q/s (all DSS)")
+        ax_rps.legend(loc="best")
+        ax_lat.set_ylabel("latency (ms)")
+        ax_lat.legend(loc="best")
+        ax_err.set_ylabel("errors (count)")
+        ax_err.set_xlabel(axis_label)
+        ax_err.legend(loc="best")
+        fig.suptitle(test_name)
+        if meta_text:
+            fig.text(
+                0.5, 0.005, meta_text, ha="center", va="bottom", fontsize=7, wrap=True
+            )
+        fig.tight_layout(rect=(0, 0.04, 1, 1))
+
+        path = outdir / f"{test_name}.png"
+        fig.savefig(path, dpi=120)
+        plt.close(fig)
+        paths.append(path)
+    return paths