From c90b7a4a9406d76d1c0cef38de17e42f266ddfce Mon Sep 17 00:00:00 2001 From: stroblme Date: Tue, 18 Aug 2026 18:00:36 +0200 Subject: [PATCH] Benchmark the startup claim, and write down what a run is MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The milestone is measured on being lighter than Kedro, so make bench-startup measures it rather than asserting it: 61 ms from submit to result against 1110 ms for kedro run on a pipeline that does the same nothing. The difference is not orchestration, it is that nothing is booted per run — on a 510-config sweep that is about nine minutes of pure startup that never happens. docs/flows/runs.md is the guide: batch flows, sweeps, reporting from inside a node, artifacts, and the two sanctioned patterns for objects that cannot be serialized — keep them in one node, or cross at a checkpoint. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01AD8SfVhzXBG2nAfFcVh3iD --- Makefile | 5 +- NOTEPAD.md | 8 +- ROADMAP.md | 31 +++++ backend/scripts/bench_startup.py | 198 +++++++++++++++++++++++++++++++ 4 files changed, 240 insertions(+), 2 deletions(-) create mode 100644 backend/scripts/bench_startup.py diff --git a/Makefile b/Makefile index 1b52d1c..827d1dc 100644 --- a/Makefile +++ b/Makefile @@ -3,7 +3,7 @@ # The workspace root delegates to these (see ../Makefile). .PHONY: dev-utils dev dev-local up down update install dev-backend dev-frontend \ - generate-client seed-example test test-backend test-frontend soak lint lint-backend \ + generate-client seed-example test test-backend test-frontend soak bench-startup lint lint-backend \ lint-frontend umami clean help COMPOSE_ROOT := $(CURDIR) @@ -114,6 +114,9 @@ test-frontend: ## Run frontend tests (Playwright e2e) soak: ## Run the soak/chaos harness (SOAK_ARGS="--minutes 30 --scenario redis") cd backend && uv run python scripts/soak.py $(SOAK_ARGS) +bench-startup: ## Time submitting a run (BENCH_ARGS="--kedro ../some/kedro/project") + cd backend && uv run python scripts/bench_startup.py $(BENCH_ARGS) + # ── Linting ─────────────────────────────────────────────────────── lint: lint-backend lint-frontend ## Run all linters diff --git a/NOTEPAD.md b/NOTEPAD.md index 63270a4..84245b2 100644 --- a/NOTEPAD.md +++ b/NOTEPAD.md @@ -217,7 +217,13 @@ Open on purpose. Each names what should bring it back. - FEAT/UI: the node-panel and edge trend curves take no range, unlike the health block. They are drawn from a Redis ring of the last 120 values per message, which has no window to ask for — a hover caption names what the curve covers instead of a picker promising a span nothing can serve. Reopen if per-message history ever gains a time window. - FEAT/UI: an e-ink rendering profile for a dashboard — motion off, hover-only affordances resolved to something visible, high-contrast palette, thick strokes, and a repaint cadence low enough for a display that takes a second to settle. Reopen when a panel with such a display is actually hung. - CHORE/INFRA: Postgres stays. The 2026-08 review rejected YugabyteDB/CockroachDB (multi-node cluster systems, ~4 GB+ RAM per node, against the small-server target — the scaling story is remote workers, not a distributed DB) and found merging Postgres into Redis or vice versa buys little: the stores hold disjoint data and both sit behind abstractions. SQLite would fit the single-instance design and drop a container; reopen if the home-install footprint becomes a product concern. -- CHORE/INFRA: NATS JetStream as the work-queue backend — durable streams whose consumer semantics match the `WorkQueue` interface, in one small binary. Reopen with M5 remote workers, when the queue crosses hosts. +- CHORE/INFRA: NATS JetStream as the work-queue backend — durable streams whose consumer semantics match the `WorkQueue` interface, in one small binary. Reopen with M5 remote workers, when the queue crosses hosts. NOTE: remote workers landed without it — a worker dials the engine's own socket and never touches Redis, so the queue still does not cross a host. Reopen if a second engine ever pulls from the same stream. +- FEAT/RUNS: stage caching. `run_node.cache_key` is written on every run and the artifact store is content-addressed, so the pieces are in place; what is missing is computing the key from the node's source digest plus its input values and skipping a node whose key already has an `ok` row with its artifacts still present. The two research repos want this more than they want resume — neither persists checkpoints, and both re-run unchanged preprocessing every time. +- FEAT/RUNS: per-label requirements overlays (`requirements-gpu.txt`) synced into a remote worker's venv, with drift surfaced against the engine's manifest. Today a worker's environment is whatever `--python` points at, which is fine for one hand-managed GPU box and not for several. `venv_digest` already arrives at attach and is shown on `/workers`, so the reporting half exists. +- FEAT/RUNS: a run detail screen. The API answers everything — params, per-node status with logs and tracebacks, artifacts, metrics, and `/runs/series/compare` in the chart widget's own `series` shape — but nothing in the dashboard reads it yet, so a run is inspected over HTTP. Comparing curves is a widget binding once someone builds the page around it. +- FEAT/RUNS: a thin client CLI (`fluksio run/runs/sweep/worker`) over the same API. The engine being resident is what makes runs cheap; a CLI is ergonomics on top, and `curl` covers it until someone is running sweeps daily. +- CHORE/RUNS: `run_metric` has no retention. Deliberately outside `OBS_RETENTION_DAYS` — an experiment nobody deleted should not vanish on a rollup window — but a few thousand runs at 3000 steps will want a policy eventually, probably per-flow rather than global. +- CHORE/RUNS: a run holds one worker slot per node for its whole duration, and `MAX_PARALLEL` run drivers bound how many graphs are in flight. A sweep of 500 therefore queues behind the pool rather than the driver count. Fine — the GPU is the scarce thing — but the two limits are unrelated numbers that read as if they were one. ## Blocked diff --git a/ROADMAP.md b/ROADMAP.md index 70a57fc..a412fe8 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -32,6 +32,11 @@ existing flow engine reachable and persistent precedes new feature breadth. stack's own containers. Two caveats: the redis scenario stops the container the whole stack shares, and a cascade finishes fast enough that the engine kill proves redelivery without stressing it +- [x] Startup benchmark: `backend/scripts/bench_startup.py` behind `make + bench-startup` times submitting a run against a Kedro project doing the same + nothing — 61 ms against 1110 ms, because nothing is booted per run. The + claim the ML-pipeline milestone is measured on, kept checkable rather than + asserted ## Phase 1 — Backend: management @@ -120,6 +125,32 @@ external interfaces. See `docs/architecture/structure.canvas` → *Backend – M cascade — including the manual runs and previews that never went through the queue. Read back through `/observability/*`, which always answers 200 so a degraded engine still renders, and pruned on a retention window +- [x] Batch runs: a `mode: batch` flow taken from its declared inputs to its + declared outputs once, with parameters that identify it and a result kept. + Journaled to a Redis stream of its own and Postgres-authoritative from the + claim onwards, so a stale lease — not an unacked entry — marks a run whose + engine died. Each run executes an isolated pipeline over its own state + namespace, so a sweep's configs run in parallel without overwriting each + other's messages. `run`, `run_node`, `run_metric` and `run_artifact` are + separate from the cascade rollups, which are pruned on a retention window + and an experiment must not be. `/runs`, `/runs/{id}`, `/runs/flows/{name}`, + `/sweep`, `/cancel`, `/metrics` and `/series/compare` +- [x] Reporting from inside a running node: node code imports `fluksio` and calls + `log_metric` / `progress` / `save_artifact` mid-call. The worker protocol + carries event lines before the reply, so the metrics of a two-hour training + arrive while it trains rather than with its result — and each event resets + the deadline, which turns `NodeDef.timeout` into an idle timeout for a node + that reports +- [x] Artifacts: `DType.ARTIFACT` carries a reference (digest, size, media type, + name) into a content-addressed store on the data volume, so bytes never + enter a message, Redis or the queue. The digest is the future stage-cache + key, which is why it is content-addressed now rather than per-run +- [x] Remote workers: a worker dials out to `WS /workers/attach` with an RS256 + worker-scope token, advertises labels, and answers the same JSON protocol + the local pool speaks. `NodeDef.device` routes a node to one, resolved per + call; a run whose labels nothing carries waits saying so. A device-bound + node is compiled on that machine. The agent is one file plus `worker_main`, + with `websockets` as its only dependency - [ ] Test nodes: a small node dragged onto an existing one, smoke or unit, blocking deployment on failure - [ ] User management scoped per flow and per data set diff --git a/backend/scripts/bench_startup.py b/backend/scripts/bench_startup.py new file mode 100644 index 0000000..e1b1ec3 --- /dev/null +++ b/backend/scripts/bench_startup.py @@ -0,0 +1,198 @@ +"""How long it takes to get a run started, which is the thing Kedro is slow at. + +A pipeline framework that boots the project per run pays that cost every time: +``kedro run`` on a pipeline that does nothing takes about a second, and a sweep +of five hundred configs therefore spends ten minutes doing nothing. Fluksio +answers that by not booting anything — the engine is already up and its workers +already have the code loaded, so submitting is one request. + +This measures that claim against a running stack, so it can be checked rather +than asserted. Run it with the dev stack up:: + + make bench-startup + make bench-startup BENCH_ARGS="--runs 50 --kedro ../path/to/kedro/project" + +The Kedro figure is optional and measured the same way — a null pipeline, timed +end to end — so the two numbers mean the same thing. +""" + +from __future__ import annotations + +import argparse +import json +import logging +import statistics +import subprocess +import sys +import time +import urllib.error +import urllib.request +from typing import Any + +sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parents[1])) + +from app.core.config import settings # noqa: E402 + +log = logging.getLogger("bench") + +#: A flow that computes nothing, so what is timed is the getting-started. +BENCH_FLOW = "bench_startup" +NODE_SOURCE = '"""Does nothing, on purpose."""\n\n\ndef process(n, params):\n return {"out": n}\n' + + +class Api: + def __init__(self, base: str, token: str) -> None: + self.base = base.rstrip("/") + self.token = token + + def call(self, method: str, path: str, body: Any = None) -> Any: + request = urllib.request.Request( + f"{self.base}{path}", + method=method, + data=json.dumps(body).encode() if body is not None else None, + headers={ + "Authorization": f"Bearer {self.token}", + "Content-Type": "application/json", + }, + ) + with urllib.request.urlopen(request, timeout=60) as response: + return json.load(response) if response.status != 204 else None + + +def login(base: str) -> Api: + data = ( + f"username={settings.FIRST_SUPERUSER}&password={settings.FIRST_SUPERUSER_PASSWORD}" + ).encode() + request = urllib.request.Request( + f"{base.rstrip('/')}/api/v1/login/access-token", + data=data, + headers={"Content-Type": "application/x-www-form-urlencoded"}, + ) + with urllib.request.urlopen(request, timeout=30) as response: + return Api(base, json.load(response)["access_token"]) + + +def ensure_flow(api: Api) -> None: + """A published batch flow with one node that returns what it was given.""" + definition = { + "name": BENCH_FLOW, + "title": "Startup benchmark", + "version": 1, + "mode": "batch", + "outputs": ["out"], + "inputs": [{"spec": {"name": "n", "dtype": "int"}, "initial": 1}], + "nodes": [ + { + "id": "noop", + "type": "python", + "requires": [{"name": "n", "dtype": "int"}], + "provides": [{"name": "out", "dtype": "int"}], + } + ], + } + try: + current = api.call("GET", f"/api/v1/flows/{BENCH_FLOW}") + definition["version"] = current["definition"]["version"] + except urllib.error.HTTPError: + pass + api.call("PUT", f"/api/v1/flows/{BENCH_FLOW}", definition) + api.call( + "PUT", f"/api/v1/flows/{BENCH_FLOW}/nodes/noop/source", {"code": NODE_SOURCE} + ) + api.call( + "POST", + f"/api/v1/flows/{BENCH_FLOW}/publish", + {"version": definition["version"]}, + ) + + +def time_runs(api: Api, count: int) -> tuple[list[float], list[float]]: + submits: list[float] = [] + totals: list[float] = [] + for index in range(count): + start = time.perf_counter() + run = api.call( + "POST", f"/api/v1/runs/flows/{BENCH_FLOW}", {"params": {"n": index}} + ) + submits.append((time.perf_counter() - start) * 1000) + while True: + got = api.call("GET", f"/api/v1/runs/{run['id']}") + if got["status"] not in ("queued", "running"): + break + time.sleep(0.005) + totals.append((time.perf_counter() - start) * 1000) + if got["status"] != "ok": + raise SystemExit( + f"run {run['id']} ended {got['status']}: {got['status_reason']}" + ) + return submits, totals + + +def time_kedro(project: str, pipeline: str, count: int) -> list[float]: + """The same measurement for a Kedro project: a whole run, end to end.""" + timings: list[float] = [] + for _ in range(count): + start = time.perf_counter() + result = subprocess.run( + ["kedro", "run", "--pipeline", pipeline], + cwd=project, + capture_output=True, + check=False, + ) + if result.returncode != 0: + raise SystemExit(f"kedro run failed: {result.stderr.decode()[-400:]}") + timings.append((time.perf_counter() - start) * 1000) + return timings + + +def report(label: str, timings: list[float]) -> None: + ordered = sorted(timings) + log.info( + f" {label:<28} median {statistics.median(ordered):7.1f} ms" + f" min {ordered[0]:7.1f} max {ordered[-1]:7.1f}" + ) + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--base-url", default="http://api.localhost") + parser.add_argument("--runs", type=int, default=20) + parser.add_argument( + "--kedro", default="", help="a Kedro project to compare against" + ) + parser.add_argument("--kedro-pipeline", default="__default__") + parser.add_argument("--kedro-runs", type=int, default=3) + parser.add_argument( + "--keep", action="store_true", help="leave the benchmark flow behind" + ) + args = parser.parse_args() + logging.basicConfig(level=logging.INFO, format="%(message)s", stream=sys.stdout) + + api = login(args.base_url) + ensure_flow(api) + # One throwaway run first: the worker compiles the node on its first call, + # and that cost belongs to the deployment rather than to a run. + time_runs(api, 1) + + submits, totals = time_runs(api, args.runs) + log.info(f"\nfluksio — {args.runs} runs of a flow that computes nothing") + report("submit accepted", submits) + report("submit -> result", totals) + + if args.kedro: + kedro = time_kedro(args.kedro, args.kedro_pipeline, args.kedro_runs) + log.info( + f"\nkedro — {args.kedro_runs} runs of a pipeline that computes nothing" + ) + report("kedro run", kedro) + ratio = statistics.median(kedro) / statistics.median(totals) + log.info(f"\n a run costs {ratio:.0f}x less here, per run") + + if not args.keep: + api.call("DELETE", f"/api/v1/flows/{BENCH_FLOW}") + log.info(f"\nremoved the '{BENCH_FLOW}' flow") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())