diff --git a/backend/fluksio/api/routes/observability.py b/backend/fluksio/api/routes/observability.py index 4959f67..f578563 100644 --- a/backend/fluksio/api/routes/observability.py +++ b/backend/fluksio/api/routes/observability.py @@ -17,6 +17,7 @@ from sqlalchemy import ColumnElement, Integer, cast, func from sqlalchemy import select as sa_select from sqlmodel import col, select +from fluksio import __version__ from fluksio.api.deps import FlowControllerDep, SessionDep, get_current_user from fluksio.api.routes.runs import elapsed_ms from fluksio.core.config import settings @@ -51,6 +52,11 @@ class HealthSummary(BaseModel): nodes: dict[str, int] queue: dict[str, Any] loop_lag: dict[str, float] + #: What this engine is running. A client ships ahead of the engine it + #: talks to — a `pip install -U` upgrades one and not the other — and a + #: route the client knows and the engine does not answers a flat 404. This + #: is what turns that into a sentence. Absent means older than this field. + version: str = "" class SeriesPoint(BaseModel): @@ -220,6 +226,7 @@ async def read_summary(request: Request, controller: FlowControllerDep) -> Any: if watchdog is not None else {"ewma": 0.0, "max_60s": 0.0} ), + version=__version__, ) diff --git a/backend/fluksio/api/routes/runs.py b/backend/fluksio/api/routes/runs.py index d150723..3cd9b46 100644 --- a/backend/fluksio/api/routes/runs.py +++ b/backend/fluksio/api/routes/runs.py @@ -391,30 +391,62 @@ def _cell(value: Any) -> Any: return json.dumps(value) +def _leaves(value: Any, prefix: str = "") -> Iterator[tuple[str, Any]]: + """Everything a record holds, by its dotted path. + + A node returning a record rather than a scalar is the ordinary shape — + the numbers arrive inside `final_metrics` — and a record in one cell is + not a column anybody can compare. Lists are left whole: a curve belongs in + the long table, not in a cell of this one. + """ + if isinstance(value, dict): + for key, inner in value.items(): + yield from _leaves(inner, f"{prefix}.{key}" if prefix else str(key)) + else: + yield prefix, value + + +def _dig(record: dict[str, Any], path: str) -> Any: + """A dotted path into a record: `final_metrics.train_loss`. + + A key with a dot in its own name is not reachable this way, which is the + price of the spelling. + """ + value: Any = record + for part in path.split("."): + if not isinstance(value, dict) or part not in value: + return None + value = value[part] + return value + + def _varying(runs: list[Run]) -> list[str]: """The inputs that differ across these runs — the axis of a sweep. - What a reader comparing arms wants as columns. Under two runs nothing can - differ, and a table of one run with none of its inputs in it is not worth - reading, so all of them are kept. + What a reader comparing arms wants as columns, compared leaf by leaf: two + configurations differing in one field give that field as a column rather + than two blobs that are not the same. Under two runs nothing can differ, + and a table of one run with none of its inputs in it is not worth reading, + so all of them are kept. """ - keys = sorted({key for run in runs for key in run.params}) + keys = sorted({path for run in runs for path, _ in _leaves(run.params)}) if len(runs) < 2: return keys return [ key for key in keys - if len({json.dumps(run.params.get(key), sort_keys=True) for run in runs}) > 1 + if len({json.dumps(_dig(run.params, key), sort_keys=True) for run in runs}) > 1 ] def _scored(runs: list[Run]) -> list[str]: - """A run's final numbers: every scalar its declared outputs carry.""" + """A run's final numbers: every number its declared outputs carry, however + deep it sits. A flag is not a number, and neither is a label.""" return sorted( { - key + path for run in runs - for key, value in run.result.items() + for path, value in _leaves(run.result) if isinstance(value, (int, float)) and not isinstance(value, bool) } ) @@ -526,6 +558,11 @@ def export_runs( by default the ones that vary across the selection, which is the sweep axis; ``params`` names them instead. ``metrics`` narrows the final numbers to a few of a run's declared outputs. + + Both take dotted paths into a record a node returned: + ``metrics=final_metrics.train_loss,test_metrics.known.perfect`` selects + three fields rather than two blobs, and the defaults reach the same + depth. """ runs = _selected(session, flow, status, group, ids, since, until) inputs = [part for part in params.split(",") if part] or _varying(runs) @@ -539,8 +576,8 @@ def export_runs( def chunks() -> Iterator[list[dict[str, Any]]]: for run in runs: row = {column: _cell(getattr(run, column)) for column in RUN_COLUMNS} - row.update((f"param.{k}", _cell(run.params.get(k))) for k in inputs) - row.update((f"metric.{k}", _cell(run.result.get(k))) for k in scores) + row.update((f"param.{k}", _cell(_dig(run.params, k))) for k in inputs) + row.update((f"metric.{k}", _cell(_dig(run.result, k))) for k in scores) yield [row] return _stream(format, "runs", columns, chunks()) diff --git a/backend/fluksio/flow/metrics.py b/backend/fluksio/flow/metrics.py index 7fc52c5..7aaa441 100644 --- a/backend/fluksio/flow/metrics.py +++ b/backend/fluksio/flow/metrics.py @@ -270,9 +270,7 @@ class MetricsCollector: ) ) - def _finish_run( - self, run: dict[str, Any], ts: float, status: str = "" - ) -> None: + def _finish_run(self, run: dict[str, Any], ts: float, status: str = "") -> None: """Close an open record. A run says how it ended; a cascade is told.""" run["finished_at"] = datetime.fromtimestamp(ts, UTC) run["duration_ms"] = round((ts - run["started_ts"]) * 1000, 2) diff --git a/backend/fluksio/main.py b/backend/fluksio/main.py index 4ef3024..2edba0c 100644 --- a/backend/fluksio/main.py +++ b/backend/fluksio/main.py @@ -12,6 +12,7 @@ from fastapi.routing import APIRoute from fluksio_worker.worker_main import ARTIFACT_DIR_ENV from starlette.middleware.cors import CORSMiddleware +from fluksio import __version__ from fluksio.api.main import api_router from fluksio.api.routes.alerts import read_config as read_alerts_config from fluksio.cloud import config as cloud_config @@ -283,6 +284,7 @@ _docs_enabled = settings.ENVIRONMENT != "production" app = FastAPI( title=settings.PROJECT_NAME, + version=__version__, openapi_url=f"{settings.API_V1_STR}/openapi.json" if _docs_enabled else None, docs_url="/docs" if _docs_enabled else None, redoc_url="/redoc" if _docs_enabled else None, diff --git a/backend/fluksio/sdk/cli.py b/backend/fluksio/sdk/cli.py index c7ce659..e920fc3 100644 --- a/backend/fluksio/sdk/cli.py +++ b/backend/fluksio/sdk/cli.py @@ -16,13 +16,14 @@ import json import pkgutil import sys import time -from collections.abc import Iterator +from collections.abc import Callable, Iterator from contextlib import contextmanager, nullcontext from pathlib import Path from typing import Any import httpx +from fluksio import __version__ from fluksio.sdk import FLOWS, Flow, SyncError from fluksio.sdk.client import ( GLOBAL_DATA_DIR, @@ -634,6 +635,10 @@ def _status_screen(client: Client) -> Any: head.append(" " + " · ".join(str(p) for p in problems), style="yellow") phrase, style = _portal_phrase(portal) head.append(" " + phrase, style=style) + # What it is running, since a client is upgraded on its own and a route + # this one knows may not be there. Absent from an engine older than the + # field itself, which is the answer in its own way. + head.append(f" v{summary.get('version') or '?'}", style="dim") counts = summary.get("flows") or {} nodes = summary.get("nodes") or {} @@ -873,7 +878,7 @@ def _selection(args: argparse.Namespace) -> dict[str, Any]: } -def _write_rows(rows: list[dict[str, Any]], fmt: str, out: str) -> int: +def _write_rows(rows: list[dict[str, Any]], fmt: str, out: str, hint: str = "") -> int: """The exported rows, in the format asked for, to a file or to stdout. The engine settles the columns over the whole selection before it sends @@ -881,7 +886,8 @@ def _write_rows(rows: list[dict[str, Any]], fmt: str, out: str) -> int: first row's are the header. """ if not rows: - print("fluksio: nothing matched, so nothing was written", file=sys.stderr) + empty = "fluksio: nothing matched, so nothing was written" + print(empty + (f". {hint}" if hint else ""), file=sys.stderr) return 0 if fmt == "parquet": try: @@ -904,44 +910,106 @@ def _write_rows(rows: list[dict[str, Any]], fmt: str, out: str) -> int: return 0 -def cmd_export_metrics(args: argparse.Namespace) -> int: - """Every selected run's numbers as one long table.""" +#: How many runs `--list` reads before giving up on finding a metric name. +#: The names belong to the flow's nodes rather than to a run, so the newest +#: one that recorded any is the whole vocabulary — the rest are for a +#: selection whose newest runs failed before they measured anything. +# ponytail: the first run with names wins; a name only an older run recorded +# is not listed. A `distinct` over the selection would be exact and is a route +# of its own. +LIST_SCAN = 10 + + +def _list_names(client: Client, args: argparse.Namespace) -> int: + """The metric names these runs carry, since a name is flow-qualified. + + `train_loss` is recorded as `train.train_loss`, and asking for the bare + one matches nothing — so this is the answer to "what would match". + """ + filters = _selection(args) + if "until" in filters: + # The history spells the same bound `before`, where it is also the + # cursor a page is taken from. + filters["before"] = filters.pop("until") + ids = args.run or [ + row["id"] for row in client.runs(flow=args.flow, limit=LIST_SCAN, **filters) + ] + for run_id in ids[:LIST_SCAN]: + names = sorted({point["name"] for point in client.metrics(run_id)}) + if names: + _say("\n".join(names)) + return 0 + _say("No metrics recorded by these runs.") + return 0 + + +def _too_old(client: Client) -> str: + """A route this client knows and the engine does not.""" + try: + version = (client.summary() or {}).get("version") or "" + except (ApiError, httpx.HTTPError): + version = "" + engine = f"the engine is {version}" if version else "the engine is older" + return ( + f"this engine has no export endpoints — {engine} and this client is " + f"{__version__}. Upgrade it: `pip install -U fluksio`" + ) + + +def _export( + args: argparse.Namespace, fetch: Callable[[Client], list[dict[str, Any]]] +) -> int: + """Both exports: fetch what was asked for, then write it.""" if args.format == "parquet" and not args.out: return _fail("--format parquet writes a file; name it with -o FILE") + hint = ( + "`--list` names the metrics these runs carry" + if getattr(args, "name", "") + else "" + ) try: with _client_for(args, retries=0) as client: - rows = client.export_metrics( - flow=args.flow, - ids=args.run, - name=args.name, - stride=args.stride, - **_selection(args), - ) + if getattr(args, "list_names", False): + return _list_names(client, args) + try: + rows = fetch(client) + except ApiError as exc: + if exc.status != 404: + raise + return _fail(_too_old(client)) except (SyncError, ApiError) as exc: return _fail(str(exc)) except httpx.HTTPError as exc: return _unreachable(exc) - return _write_rows(rows, args.format, args.out) + return _write_rows(rows, args.format, args.out, hint) + + +def cmd_export_metrics(args: argparse.Namespace) -> int: + """Every selected run's numbers as one long table.""" + return _export( + args, + lambda client: client.export_metrics( + flow=args.flow, + ids=args.run, + name=args.name, + stride=args.stride, + **_selection(args), + ), + ) def cmd_export_runs(args: argparse.Namespace) -> int: """One row per run: its inputs, its final numbers, what it ran.""" - if args.format == "parquet" and not args.out: - return _fail("--format parquet writes a file; name it with -o FILE") - try: - with _client_for(args, retries=0) as client: - rows = client.export_runs( - flow=args.flow, - ids=args.run, - params=args.params, - metrics=args.metrics, - **_selection(args), - ) - except (SyncError, ApiError) as exc: - return _fail(str(exc)) - except httpx.HTTPError as exc: - return _unreachable(exc) - return _write_rows(rows, args.format, args.out) + return _export( + args, + lambda client: client.export_runs( + flow=args.flow, + ids=args.run, + params=args.params, + metrics=args.metrics, + **_selection(args), + ), + ) # --------------------------------------------------------------------------- @@ -1148,6 +1216,12 @@ def add_parsers(subparsers: Any) -> None: default=1, help="keep every Nth point of each curve", ) + sub.add_argument( + "--list", + dest="list_names", + action="store_true", + help="print the metric names these runs carry, and stop", + ) sub.set_defaults(func=cmd_export_metrics) sub = exports.add_parser( @@ -1158,9 +1232,15 @@ def add_parsers(subparsers: Any) -> None: "--params", default="", metavar="A,B", - help="the inputs to put in columns (default: the ones that vary)", + help=( + "the inputs to put in columns, dotted into a record " + "(default: the ones that vary)" + ), ) sub.add_argument( - "--metrics", default="", metavar="A,B", help="the final numbers to keep" + "--metrics", + default="", + metavar="A,B", + help="the final numbers to keep, e.g. final_metrics.train_loss", ) sub.set_defaults(func=cmd_export_runs) diff --git a/backend/pyproject.toml b/backend/pyproject.toml index 3ea0651..179d12d 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "fluksio" -version = "0.1.4" +version = "0.1.5" description = "Node-based automation engine: flows, dashboards, batch runs" readme = "README.md" license = "AGPL-3.0-or-later" diff --git a/backend/tests/api/routes/test_runs.py b/backend/tests/api/routes/test_runs.py index d3e32f6..9fe41ba 100644 --- a/backend/tests/api/routes/test_runs.py +++ b/backend/tests/api/routes/test_runs.py @@ -654,7 +654,11 @@ def test_a_tree_that_did_not_move_is_not_written_again(): @pytest.fixture def exported(): - """Two runs of one flow, each with two curves and one input that varies.""" + """Two runs of one flow: two curves each, and records on both sides. + + The numbers a node returns are usually inside a record rather than at the + top of the result, so the fixture is shaped the way a real one is. + """ made = datetime.now(UTC) with Session(db_engine) as session: for index, (lr, acc) in enumerate([(0.1, 0.5), (0.01, 0.25)]): @@ -664,8 +668,17 @@ def exported(): id=run_id, flow="export-study", status="ok", - params={"lr": lr, "epochs": 10}, - result={"acc": acc, "note": "n/a"}, + params={ + "lr": lr, + "epochs": 10, + "config": {"model": "mlp", "depth": index + 1}, + }, + result={ + "acc": acc, + "note": "n/a", + "final_metrics": {"train_loss": acc * 2}, + "test_metrics": {"known": {"perfect": 1.0}}, + }, created_at=made + timedelta(seconds=index), ) ) @@ -727,17 +740,32 @@ def test_an_exported_run_row_carries_the_inputs_that_vary( rows = _lines(export(format="jsonl")) assert [row["id"] for row in rows] == ["exp-1", "exp-0"] assert {row["param.lr"] for row in rows} == {0.1, 0.01} - # `epochs` is the same on both runs, so it is not what they differ by; a - # string in the result is not one of the run's numbers. + # `epochs` is the same on both runs, so it is not what they differ by, and + # neither is the model inside the config — but the depth beside it is. assert "param.epochs" not in rows[0] - assert rows[0]["metric.acc"] == 0.25 - assert "metric.note" not in rows[0] + assert "param.config.model" not in rows[0] + assert {row["param.config.depth"] for row in rows} == {1, 2} assert "param.epochs" in _lines(export(format="jsonl", params="epochs"))[0] + # A number inside a record is a column of its own, however deep; a string + # is not one of the run's numbers wherever it sits. + assert rows[0]["metric.acc"] == 0.25 + assert rows[0]["metric.final_metrics.train_loss"] == 0.5 + assert rows[0]["metric.test_metrics.known.perfect"] == 1.0 + assert "metric.note" not in rows[0] + + named = _lines(export(format="jsonl", metrics="test_metrics.known.perfect")) + assert list(named[0])[-1] == "metric.test_metrics.known.perfect" + assert "metric.acc" not in named[0] + # csv is the default, and the columns are in the order the header names. header = export().text.splitlines()[0] assert header.startswith("id,flow,status,") - assert header.endswith("param.lr,metric.acc") + assert header.endswith( + "param.config.depth,param.lr," + "metric.acc,metric.final_metrics.train_loss," + "metric.test_metrics.known.perfect" + ) @pytest.fixture diff --git a/backend/tests/test_cli.py b/backend/tests/test_cli.py index aa7edbd..62e65c5 100644 --- a/backend/tests/test_cli.py +++ b/backend/tests/test_cli.py @@ -345,3 +345,46 @@ def test_an_export_is_parsed_with_its_selection() -> None: assert runs.func is cmd_export_runs assert runs.params == "lr" assert runs.format == "jsonl" + + +def test_the_metric_names_are_asked_for_rather_than_guessed() -> None: + """A name is flow-qualified, so `--list` is what says what would match.""" + from fluksio.cli import _parser + from fluksio.sdk.cli import _list_names + + parser = _parser() + assert parser.parse_args(["export", "metrics"]).list_names is False + assert parser.parse_args(["export", "metrics", "--list"]).list_names is True + + class Engine: + def runs(self, flow="", limit=0, **filters): + assert filters == {"status": "ok"} + return [{"id": "r-empty"}, {"id": "r-1"}] + + def metrics(self, run_id, name="", stride=1): + # The newest run failed before it measured anything; the next one + # carries the vocabulary. + return [] if run_id == "r-empty" else [{"name": "train.train_loss"}] + + args = parser.parse_args( + ["export", "metrics", "--flow", "train", "--status", "ok", "--list"] + ) + assert _list_names(Engine(), args) == 0 + + +def test_an_engine_without_the_route_is_named_rather_than_404() -> None: + """A client ships ahead of the engine; a flat 404 does not say so.""" + from fluksio.sdk.cli import _too_old + + class Old: + def summary(self): + return {"status": "ok", "version": "0.1.4"} + + class Ancient: + def summary(self): + # Older than the field itself. + return {"status": "ok"} + + assert "engine is 0.1.4" in _too_old(Old()) + assert "engine is older" in _too_old(Ancient()) + assert "pip install -U fluksio" in _too_old(Ancient()) diff --git a/docs/code/api.md b/docs/code/api.md index 48b3161..7ada624 100644 --- a/docs/code/api.md +++ b/docs/code/api.md @@ -129,7 +129,9 @@ Both exports stream `csv` (the default) or `jsonl`, and take the selection the history takes plus `?ids=a,b,c`, `?since=` and `?until=`. `export/runs` puts the inputs that *vary* across the selection in `param.` columns — the sweep axis — unless `?params=` names them, and the run's numbers in `metric.` -columns. The run id is on every row of both, which is what makes an exported +columns. Both are dotted paths into whatever a node returned, so +`?metrics=final_metrics.train_loss` selects a field of a record and the +defaults reach every number inside one. The run id is on every row of both, which is what makes an exported file a join back to the run rather than a loose table. `compare` answers in the same shape a chart widget draws, so three training @@ -186,7 +188,7 @@ read one back. | Path | What | |---|---| -| `/observability/summary` | engine health — always 200, degraded or not | +| `/observability/summary` | engine health — always 200, degraded or not — and the `version` it is running | | `/observability/timeseries` | executions and failures over a window | | `/observability/flows` | per-flow rollups with a 60-slice trend | | `/observability/runs` | recent cascades, with `?flow=`, `?since=`, `?until=` | diff --git a/docs/code/cli.md b/docs/code/cli.md index 5700a17..c954665 100644 --- a/docs/code/cli.md +++ b/docs/code/cli.md @@ -307,7 +307,8 @@ of them are finished and exits non-zero if any failed. ### `fluksio export` ```sh -fluksio export metrics --flow train --name train_loss,val_loss --stride 10 -o curves.csv +fluksio export metrics --flow train --list +fluksio export metrics --flow train --name train.train_loss --stride 10 -o curves.csv fluksio export runs --flow train --status ok > arms.csv ``` @@ -323,6 +324,18 @@ runs — the axis of the sweep, which is what a comparison is read along — unless `--params lr,seed` names them. `--metrics` narrows the final numbers the same way. +A node usually returns a record rather than a scalar, so both sides take +dotted paths into one: `--metrics final_metrics.train_loss,test_metrics.known.perfect` +selects three fields rather than two blobs, and `--params model.ansatz` does +the same for an input. The defaults reach the same depth — every number a +result carries becomes a column wherever it sits, and inputs are compared +leaf by leaf, so two configurations differing in one field give that field +rather than two records that are merely not equal. + +Metric names are flow-qualified — a node of `train` writing `train_loss` +records `train.train_loss` — so `--list` prints the names the selected runs +carry when the spelling is not obvious. + Both take `--flow`, `--run ID` (repeat it), `--group`, `--status`, `--since`, `--until` and `--local`, and both put the run id on every row: it is the join back to the run page and to what the run made. diff --git a/docs/concepts/runs.md b/docs/concepts/runs.md index d00289e..0cfa433 100644 --- a/docs/concepts/runs.md +++ b/docs/concepts/runs.md @@ -340,6 +340,12 @@ and the code digest — an exported file says what produced its numbers. [The CLI](../code/cli.md#fluksio-export) writes the same rows as csv, jsonl or parquet, which is where an export belongs: in the script beside the analysis. +A node's numbers usually arrive inside a record, and the wide table follows +them in: every number a result carries is a column of its own, named by its +path — `metric.final_metrics.train_loss` — and `metrics="final_metrics.train_loss"` +selects one. Inputs work the same way, and are compared leaf by leaf, so two +configurations differing in one field give that field as the axis. + ### A dashboard, read against runs A run records values under the same names a dashboard binds to — a run of diff --git a/docs/getting-started/data-science.md b/docs/getting-started/data-science.md index 2a7cf88..4bf66e8 100644 --- a/docs/getting-started/data-science.md +++ b/docs/getting-started/data-science.md @@ -618,8 +618,12 @@ fluksio export runs --flow train --status ok -o arms.csv Both carry the run id on every row, and `arms.csv` carries the commit and the code digest beside it, so an exported file still says what produced its -numbers. `Client.export_metrics()` and `Client.export_runs()` answer the same -rows to a notebook, ready for `pandas.DataFrame`. +numbers. Numbers inside a record are columns of their own — +`metric.final_metrics.train_loss` — and `--metrics` and `--params` take those +dotted paths to narrow the table. Metric names are flow-qualified, so +`--list` prints the ones a selection carries. `Client.export_metrics()` and +`Client.export_runs()` answer the same rows to a notebook, ready for +`pandas.DataFrame`. ## Looking at them in the portal diff --git a/uv.lock b/uv.lock index 3a17484..3130bf5 100644 --- a/uv.lock +++ b/uv.lock @@ -869,7 +869,7 @@ wheels = [ [[package]] name = "fluksio" -version = "0.1.4" +version = "0.1.5" source = { editable = "backend" } dependencies = [ { name = "aiomqtt" },