diff --git a/backend/fluksio/api/routes/flavors.py b/backend/fluksio/api/routes/flavors.py index 133ee0e..71a974a 100644 --- a/backend/fluksio/api/routes/flavors.py +++ b/backend/fluksio/api/routes/flavors.py @@ -110,19 +110,20 @@ async def delete_flavor( def _usages(controller: FlowControllerDep, name: str) -> list[str]: - """Which nodes name this flavor, published or in a draft.""" + """Which nodes name this flavor, published or in a draft, as + ``.``.""" used: list[str] = [] - for flow_name in controller.store.list_flows(): + for flow_id in controller.store.list_flows(): seen = set() for draft in (False, True): - if draft and not controller.store.has_draft(flow_name): + if draft and not controller.store.has_draft(flow_id): continue try: - flow = controller.store.read_flow(flow_name, draft=draft) + flow = controller.store.read_flow(flow_id, draft=draft) except Exception: continue for node in flow.nodes: if node.resources is not None and node.resources.flavor == name: - seen.add(f"{flow_name}.{node.id}") + seen.add(f"{flow.name}.{node.id}") used.extend(sorted(seen)) return used diff --git a/backend/fluksio/api/routes/flows.py b/backend/fluksio/api/routes/flows.py index fd74bea..082d1f4 100644 --- a/backend/fluksio/api/routes/flows.py +++ b/backend/fluksio/api/routes/flows.py @@ -119,6 +119,9 @@ class Endpoint(BaseModel): kind: str #: Stable within its kind, and used as the canvas node id. id: str + #: The id of the dashboard or flow it sits in, which is what opening it + #: goes by. Never read it off ``id``, whose flow part is a key. + owner_id: str = "" label: str #: What sort of widget, or the node type in the other flow. detail: str = "" @@ -194,24 +197,31 @@ def _endpoints(controller: FlowController, definition: FlowDef) -> list[Endpoint found[f"dashboard:{binding['dashboard']}:{binding['widget']}"] = Endpoint( kind="dashboard", id=f"dashboard:{binding['dashboard']}:{binding['widget']}", + owner_id=binding["dashboard"], label=binding["title"], detail=binding["type"], provides=[binding["provides"]] if binding["provides"] else [], requires=binding["requires"], ) - def _flow_endpoint(key: str, detail: str) -> Endpoint: + def _flow_endpoint(key: str, detail: str, owner_id: str = "") -> Endpoint: """One label per node on the far side, however many names reach it.""" return found.setdefault( f"flow:{key}", - Endpoint(kind="flow", id=f"flow:{key}", label=key, detail=detail), + Endpoint( + kind="flow", + id=f"flow:{key}", + owner_id=owner_id, + label=key, + detail=detail, + ), ) # An outsider reaching into this flow. - for other, node_id, provides, requires in controller.cross_flow_nodes(flow): - endpoint = _flow_endpoint(f"{other}.{node_id}", "flow") - endpoint.provides += provides - endpoint.requires += requires + for other in controller.cross_flow_nodes(flow): + endpoint = _flow_endpoint(f"{other.flow}.{other.node}", "flow", other.flow_id) + endpoint.provides += other.provides + endpoint.requires += other.requires # And this flow reaching out: its own ports bound to a message of another # flow. Read from the working document rather than from the store, so a @@ -235,6 +245,7 @@ def _endpoints(controller: FlowController, definition: FlowDef) -> list[Endpoint endpoint = _flow_endpoint( f"{far.flow}.{far.node}" if far else message, far.type if far else "flow", + far.flow_id if far else "", ) # An endpoint publishing into this flow is what this flow reads. side = endpoint.requires if ours_publishes else endpoint.provides @@ -289,14 +300,15 @@ def _read_flow(controller: FlowController, flow_id: str) -> FlowDef: raise _missing(flow_id) -def _audit(action: str, flow: str, user: CurrentUser) -> None: - """Record who changed what, by the flow's key. The collector writes it - down; the bus carries it.""" +def _audit(action: str, key: str, flow_id: str, user: CurrentUser) -> None: + """Record who changed what. The collector writes it down under the flow's + id; the bus carries the key with it, as it does for every event.""" event_bus.publish( { "type": "audit", "action": action, - "flow": flow, + "flow": key, + "flow_id": flow_id, "user": user.email, "ts": time.time(), } @@ -415,13 +427,12 @@ async def delete_shared_node( first, so the source can be retired while flows still use it. """ if detach: - running = await run_in_threadpool(controller.store.usages, lib_name, False) + running = await run_in_threadpool(controller.store.lib_users, lib_name) try: await run_in_threadpool(controller.store.detach_lib, lib_name) except LibNotFound: raise HTTPException(status_code=404, detail=f"No shared node '{lib_name}'") - # `usages` qualifies with the flow's id. - for flow_id in sorted({usage.split(".", 1)[0] for usage in running}): + for flow_id in running: await controller.reload_flow(flow_id) return Message(message=f"Deleted shared node '{lib_name}'") # The published flows too: a draft that unshared a node still runs it. @@ -643,7 +654,7 @@ async def publish_flow( status_code=409, detail={"message": str(exc), "current_version": exc.current}, ) - _audit("published", definition.name, user) + _audit("published", definition.name, flow_id, user) await controller.reload_flow(flow_id) return await run_in_threadpool(_detail, controller, published) @@ -680,7 +691,7 @@ async def delete_flow( """ live = session.exec( select(Run.id).where( - col(Run.flow) == flow_id, col(Run.status).in_(("running", "queued")) + col(Run.flow_id) == flow_id, col(Run.status).in_(("running", "queued")) ) ).first() if live is not None: @@ -695,7 +706,7 @@ async def delete_flow( key = await run_in_threadpool(controller.store.delete_flow, flow_id) except FlowNotFound: raise _missing(flow_id) - _audit("deleted", key, user) + _audit("deleted", key, flow_id, user) # Its nodes first, or they would put values and work straight back. # Its runs stay: a run record is a record of what ran, and surviving the # flow it belonged to is the point of keeping one. `DELETE /runs?flow=` @@ -824,7 +835,7 @@ async def save_node_source( controller.store.write_lib_source, ref, source.code ) if changed: - _audit(f"edited shared node '{ref}'", name, user) + _audit(f"edited shared node '{ref}'", name, flow_id, user) await controller.reload_lib_users(ref) else: await run_in_threadpool( @@ -905,7 +916,7 @@ async def start_flow( """Let the engine run this flow again.""" definition = _read_flow(controller, flow_id) await controller.set_enabled(flow_id, definition.name, True) - _audit("started", definition.name, user) + _audit("started", definition.name, flow_id, user) return await run_in_threadpool(_detail, controller, definition) @@ -916,7 +927,7 @@ async def stop_flow( """Take this flow off the engine: no subscriptions, schedules or webhooks.""" definition = _read_flow(controller, flow_id) await controller.set_enabled(flow_id, definition.name, False) - _audit("stopped", definition.name, user) + _audit("stopped", definition.name, flow_id, user) return await run_in_threadpool(_detail, controller, definition) @@ -1123,9 +1134,14 @@ async def train_node( if service is None: raise HTTPException(status_code=503, detail="Runs are not available") try: + # Registered by `fluksio sync`, which gives it whatever id it gets: the + # built-in flow is known by its key. + train_id = controller.store.key_index().get(TRAIN_FLOW) + if train_id is None: + raise FlowNotFound(TRAIN_FLOW) run = await run_in_threadpool( service.submit, - TRAIN_FLOW, + train_id, params={"installation": orjson.dumps(wren.installation()).decode()}, cause="train", actor=user.email, @@ -1153,11 +1169,15 @@ def read_training( controller: FlowControllerDep, session: SessionDep, ) -> Any: - """A wren node's latest training run, or null when it has none.""" + """A wren node's latest training run, or null when it has none. + + Found by the node it trains alone: only a training run names one, and + the training flow may have been registered again under another id since. + """ _node_def(controller, flow_id, node_id) run = session.exec( select(Run) - .where(col(Run.flow) == TRAIN_FLOW, col(Run.for_node) == f"{flow_id}.{node_id}") + .where(col(Run.for_node) == f"{flow_id}.{node_id}") .order_by(col(Run.created_at).desc()) ).first() if run is None: diff --git a/backend/fluksio/api/routes/observability.py b/backend/fluksio/api/routes/observability.py index c42961d..4c1e913 100644 --- a/backend/fluksio/api/routes/observability.py +++ b/backend/fluksio/api/routes/observability.py @@ -28,6 +28,7 @@ from fluksio.api.deps import ( from fluksio.api.routes.runs import elapsed_ms from fluksio.core.config import settings from fluksio.flow.controller import NodeStatus +from fluksio.flow.tombstones import names_of, shown_nodes, stored_node from fluksio.models import EngineEvent, FlowRun, Message, MetricBucket router = APIRouter( @@ -77,7 +78,9 @@ class SeriesPoint(BaseModel): class FlowRollup(BaseModel): + #: The flow's id, and its key: what it is called now, or was last called. flow: str + flow_name: str = "" executions: int errors: int messages: int @@ -100,7 +103,9 @@ class NodeTypeTiming(BaseModel): class RunRow(BaseModel): id: str + #: The flow's id, and its key as `FlowRollup` has it. flow: str + flow_name: str = "" source: str #: The run this one continues: for a flush, the run whose value it released. parent: str = "" @@ -130,7 +135,11 @@ class EventRow(BaseModel): id: int ts: datetime type: str + #: The flow's id, and its key as `FlowRollup` has it. Both empty for the + #: engine as a whole. flow: str + flow_name: str = "" + #: ``.``, or empty. node: str detail: str actor: str @@ -138,6 +147,8 @@ class EventRow(BaseModel): class DeadLetter(BaseModel): + """A work item given up on, named as the engine runs it: by key.""" + id: str ts: float flow: str @@ -317,6 +328,7 @@ async def read_summary(request: Request, controller: FlowControllerDep) -> Any: @router.get("/timeseries", response_model=list[SeriesPoint]) def read_timeseries( session: SessionDep, + controller: FlowControllerDep, flow: str | None = None, node: str | None = None, hours: Annotated[int, Query(ge=0, le=8760)] = 24, @@ -331,6 +343,8 @@ def read_timeseries( the same pair ``/runs`` and ``/events`` take — so a chart dragged to a span can be re-fetched at that span's own resolution rather than magnifying the buckets it already has. + + ``flow`` is a flow id, ``node`` a ``.`` as the engine names it. """ hours = _window_hours(hours) # The database does the fold: a week of minute rows per node used to cross @@ -356,9 +370,10 @@ def read_timeseries( col(MetricBucket.bucket) < (_aware(until) if until else _closed_before(stride)), ) if flow: - statement = statement.where(col(MetricBucket.flow) == flow) + statement = statement.where(col(MetricBucket.flow_id) == flow) if node: - statement = statement.where(col(MetricBucket.node) == node) + stored = stored_node(node, controller.store, flow or "") + statement = statement.where(col(MetricBucket.node) == stored) return [ SeriesPoint( @@ -376,7 +391,9 @@ def read_timeseries( @router.get("/flows", response_model=list[FlowRollup]) def read_flow_rollups( - session: SessionDep, hours: Annotated[int, Query(ge=0, le=8760)] = 24 + session: SessionDep, + controller: FlowControllerDep, + hours: Annotated[int, Query(ge=0, le=8760)] = 24, ) -> Any: """One row per flow, with a coarse trend of how much it ran.""" hours = _window_hours(hours) @@ -399,7 +416,7 @@ def read_flow_rollups( ).label("slot") statement = ( sa_select( - col(MetricBucket.flow), + col(MetricBucket.flow_id), slot, func.sum(col(MetricBucket.executions)).label("executions"), func.sum(col(MetricBucket.errors)).label("errors"), @@ -412,13 +429,13 @@ def read_flow_rollups( col(MetricBucket.bucket) >= since, col(MetricBucket.bucket) < datetime.fromtimestamp(float(end), UTC), ) - .group_by(col(MetricBucket.flow), slot) + .group_by(col(MetricBucket.flow_id), slot) ) rollups: dict[str, dict[str, Any]] = {} for row in session.execute(statement): entry = rollups.setdefault( - row.flow, + row.flow_id, { "executions": 0, "errors": 0, @@ -448,15 +465,17 @@ def read_flow_rollups( # mapping and subscript it. last_errors = dict( session.exec( - select(col(EngineEvent.flow), func.max(col(EngineEvent.ts))) + select(col(EngineEvent.flow_id), func.max(col(EngineEvent.ts))) .where(col(EngineEvent.type) == "node_error", col(EngineEvent.ts) >= since) - .group_by(col(EngineEvent.flow)) + .group_by(col(EngineEvent.flow_id)) ).all() ) + names = names_of(rollups, controller.store) return [ FlowRollup( flow=flow, + flow_name=names[flow], executions=entry["executions"], errors=entry["errors"], messages=entry["messages"], @@ -483,8 +502,9 @@ def read_node_type_timings( flows. A node deleted since it ran has no type left to count under. """ hours = _window_hours(hours) + # By `.`, the way the rollups name a node. types = { - f"{flow.name}.{node.id}": node.type + f"{flow.id}.{node.id}": node.type for flow in controller.store.read_all() for node in flow.nodes } @@ -526,6 +546,7 @@ def read_node_type_timings( @router.get("/runs", response_model=RunPage) def read_runs( session: SessionDep, + controller: FlowControllerDep, flow: str | None = None, status: str | None = None, since: datetime | None = None, @@ -536,10 +557,11 @@ def read_runs( ``since`` is inclusive and ``until`` exclusive, so a window of one minute holds exactly the runs of the minute bucket the charts are drawn from. + ``flow`` is a flow id. """ filters: list[ColumnElement[bool]] = [] if flow: - filters.append(col(FlowRun.flow) == flow) + filters.append(col(FlowRun.flow_id) == flow) if status: filters.append(col(FlowRun.status) == status) if since: @@ -563,12 +585,20 @@ def read_runs( ).one() ) ) - return {"data": rows, "count": count} + names = names_of({row.flow_id for row in rows}, controller.store) + data = [ + RunRow.model_validate( + {**row.model_dump(), "flow": row.flow_id, "flow_name": names[row.flow_id]} + ) + for row in rows + ] + return {"data": data, "count": count} @router.get("/events", response_model=list[EventRow]) def read_events( session: SessionDep, + controller: FlowControllerDep, kind: Literal["failure", "audit"] = "failure", flow: str | None = None, run: str | None = None, @@ -582,7 +612,8 @@ def read_events( takes, so a list can cover the span the charts beside it are drawn from. ``run`` narrows to one run — a batch run's id, or the journaled item a live cascade came from. Rows recorded before the column existed carry none, so - an old failure answers no run at all rather than the wrong one. + an old failure answers no run at all rather than the wrong one. ``flow`` + is a flow id. """ statement = select(EngineEvent).order_by(col(EngineEvent.ts).desc()) if kind == "audit": @@ -590,14 +621,28 @@ def read_events( else: statement = statement.where(col(EngineEvent.type) != "audit") if flow: - statement = statement.where(col(EngineEvent.flow) == flow) + statement = statement.where(col(EngineEvent.flow_id) == flow) if run: statement = statement.where(col(EngineEvent.run) == run) if since: statement = statement.where(col(EngineEvent.ts) >= _aware(since)) if until: statement = statement.where(col(EngineEvent.ts) < _aware(until)) - return list(session.exec(statement.limit(min(limit, 500)))) + rows = list(session.exec(statement.limit(min(limit, 500)))) + store = controller.store + names = names_of({row.flow_id for row in rows if row.flow_id}, store) + nodes = shown_nodes({row.node for row in rows if row.node}, store) + return [ + EventRow.model_validate( + { + **row.model_dump(), + "flow": row.flow_id, + "flow_name": names.get(row.flow_id, ""), + "node": nodes.get(row.node, ""), + } + ) + for row in rows + ] @router.get("/dead-letter", response_model=list[DeadLetter]) diff --git a/backend/fluksio/api/routes/runs.py b/backend/fluksio/api/routes/runs.py index c895e9b..f1d1a22 100644 --- a/backend/fluksio/api/routes/runs.py +++ b/backend/fluksio/api/routes/runs.py @@ -22,13 +22,19 @@ from sqlalchemy import delete, func from sqlalchemy import select as sa_select from sqlmodel import Session, col, select -from fluksio.api.deps import CurrentUser, SessionDep, get_current_user +from fluksio.api.deps import ( + CurrentUser, + FlowControllerDep, + SessionDep, + get_current_user, +) from fluksio.core.config import settings from fluksio.core.db import writing from fluksio.flow.events import event_bus from fluksio.flow.messages import requalify from fluksio.flow.runs import OUTPUT_CAP, RunRejected, RunService, new_run_id -from fluksio.flow.store import FlowNotFound +from fluksio.flow.store import FlowNotFound, FlowStore +from fluksio.flow.tombstones import name_of, names_of, shown_nodes, stored_node from fluksio.models import FlowRun, Message, Run, RunArtifact, RunMetric, RunNode router = APIRouter( @@ -105,6 +111,7 @@ class ImportArtifact(BaseModel): class ImportEntry(BaseModel): """One run executed somewhere else, as it ended there.""" + #: The flow's id. flow: str #: What it was called where it ran. Importing it again is a no-op. external_id: str = Field(min_length=1, max_length=255) @@ -133,6 +140,7 @@ class Imported(BaseModel): class RunNodeRow(BaseModel): + #: ``.``, the way the engine names it. node: str status: str attempt: int @@ -153,6 +161,7 @@ class RunNodeRow(BaseModel): class ArtifactRow(BaseModel): name: str + #: ``.``. node: str digest: str size: int @@ -167,7 +176,11 @@ class RunRow(BaseModel): """A run without its result, which is the part that can be large.""" id: str + #: The flow's id, which the history is kept under. flow: str + #: Its key: what it is called now, or what it was last called once it is + #: deleted. The id where nothing remembers either. + flow_name: str = "" status: str status_reason: str cause: str @@ -224,7 +237,9 @@ class RunDetail(RunRow): class FlowRunsRow(BaseModel): """How much a flow has been run, for the screen's list of flows.""" + #: The flow's id, and its key as `RunRow` has it. flow: str + flow_name: str = "" runs: int running: int queued: int @@ -260,16 +275,36 @@ def _service(request: Request) -> RunService: return service -@router.post("/flows/{name}", response_model=RunRow, status_code=202) +def _named[Row: RunRow]( + model: type[Row], runs: list[Run], store: FlowStore +) -> list[Row]: + """Runs as rows, ``flow`` their flow's id and ``flow_name`` its key. + + One lookup for the lot, however many flows the page holds. + """ + names = names_of({run.flow_id for run in runs}, store) + return [ + model.model_validate( + {**run.model_dump(), "flow": run.flow_id, "flow_name": names[run.flow_id]} + ) + for run in runs + ] + + +@router.post("/flows/{flow_id}", response_model=RunRow, status_code=202) async def create_run( - name: str, body: RunCreate, request: Request, user: CurrentUser + flow_id: str, + body: RunCreate, + request: Request, + user: CurrentUser, + controller: FlowControllerDep, ) -> Any: """Queue one run of a flow.""" service = _service(request) try: - return await run_in_threadpool( + run = await run_in_threadpool( service.submit, - name, + flow_id, params=body.params, seed=body.seed, cause=body.cause, @@ -282,11 +317,16 @@ async def create_run( raise HTTPException(status_code=404, detail=str(exc)) from exc except RunRejected as exc: raise HTTPException(status_code=422, detail=str(exc)) from exc + return _named(RunRow, [run], controller.store)[0] -@router.post("/flows/{name}/sweep", response_model=list[RunRow], status_code=202) +@router.post("/flows/{flow_id}/sweep", response_model=list[RunRow], status_code=202) async def create_sweep( - name: str, body: SweepCreate, request: Request, user: CurrentUser + flow_id: str, + body: SweepCreate, + request: Request, + user: CurrentUser, + controller: FlowControllerDep, ) -> Any: """Queue many runs of one flow under a shared group. @@ -306,7 +346,7 @@ async def create_sweep( def submit_all() -> list[Run]: return [ service.submit( - name, + flow_id, params=entry.params, seed=entry.seed, group_id=group, @@ -320,11 +360,12 @@ async def create_sweep( ] try: - return await run_in_threadpool(submit_all) + runs = await run_in_threadpool(submit_all) except FlowNotFound as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc except RunRejected as exc: raise HTTPException(status_code=422, detail=str(exc)) from exc + return _named(RunRow, runs, controller.store) @router.post("/import", response_model=list[Imported], status_code=202) @@ -369,6 +410,7 @@ async def import_runs(body: ImportCreate, request: Request, user: CurrentUser) - @router.get("", response_model=list[RunRow]) def read_runs( session: SessionDep, + controller: FlowControllerDep, flow: str | None = None, status: str | None = None, group: str | None = None, @@ -384,11 +426,11 @@ def read_runs( first, so handing back the last row's ``created_at`` reads the next page whatever landed meanwhile — which ``offset`` cannot, since a run submitted between two pages shifts every row down one. ``since`` bounds the other - end and is inclusive. + end and is inclusive. ``flow`` is a flow id. """ statement = select(Run).order_by(col(Run.created_at).desc()) if flow: - statement = statement.where(col(Run.flow) == flow) + statement = statement.where(col(Run.flow_id) == flow) if status: statement = statement.where(col(Run.status) == status) if group: @@ -400,11 +442,11 @@ def read_runs( if before: statement = statement.where(col(Run.created_at) < _aware(before)) statement = statement.offset(max(0, offset)).limit(min(limit, 500)) - return list(session.exec(statement)) + return _named(RunRow, list(session.exec(statement)), controller.store) @router.get("/overview", response_model=list[FlowRunsRow]) -def read_overview(session: SessionDep) -> Any: +def read_overview(session: SessionDep, controller: FlowControllerDep) -> Any: """One row per flow that has ever run, busiest-recent first. The list caps at 500 newest runs, so counting flows on the client goes @@ -416,11 +458,11 @@ def read_overview(session: SessionDep) -> Any: # rather than making it cheaper to find. A rollup is the answer if the # scan ever measures. statement = sa_select( - col(Run.flow), + col(Run.flow_id), col(Run.status), func.count(col(Run.id)), func.max(col(Run.created_at)), - ).group_by(col(Run.flow), col(Run.status)) + ).group_by(col(Run.flow_id), col(Run.status)) rows: dict[str, FlowRunsRow] = {} for flow, status, count, latest in session.execute(statement): @@ -432,6 +474,9 @@ def read_overview(session: SessionDep) -> Any: row.queued += count if row.last_created_at is None or latest > row.last_created_at: row.last_created_at = latest + names = names_of(rows, controller.store) + for flow_id, row in rows.items(): + row.flow_name = names[flow_id] return sorted(rows.values(), key=lambda row: row.last_created_at, reverse=True) @@ -463,7 +508,9 @@ EXPORT_CAP = 10_000 RUN_COLUMNS = ( "id", + #: The flow's id, and its key as `RunRow` has it. "flow", + "flow_name", "status", "created_at", "started_at", @@ -489,14 +536,14 @@ def _selection( since: datetime | None, until: datetime | None, ) -> Any: - """The query behind a selection of runs, newest first. + """The query behind a selection of runs, newest first. ``flow`` is an id. Separate from reading it, because the names route asks the same question as a subquery rather than for the rows. """ statement = select(Run).order_by(col(Run.created_at).desc()) if flow: - statement = statement.where(col(Run.flow) == flow) + statement = statement.where(col(Run.flow_id) == flow) if status: statement = statement.where(col(Run.status) == status) if group: @@ -655,6 +702,7 @@ def _stream( @router.get("/export/metrics") def export_metrics( session: SessionDep, + controller: FlowControllerDep, flow: str | None = None, status: str | None = None, group: str | None = None, @@ -679,7 +727,7 @@ def export_metrics( for run in runs: rows: list[dict[str, Any]] = [] for series, points in groupby( - _series(session, run.id), key=lambda row: row.name + _series(session, controller.store, run.id), key=lambda row: row.name ): if wanted and series not in wanted: continue @@ -701,6 +749,7 @@ def export_metrics( @router.get("/export/runs") def export_runs( session: SessionDep, + controller: FlowControllerDep, flow: str | None = None, status: str | None = None, group: str | None = None, @@ -734,9 +783,15 @@ def export_runs( *(f"metric.{key}" for key in scores), ] + names = names_of({run.flow_id for run in runs}, controller.store) + def chunks() -> Iterator[list[dict[str, Any]]]: for run in runs: - row = {column: _cell(getattr(run, column)) for column in RUN_COLUMNS} + own = {"flow": run.flow_id, "flow_name": names[run.flow_id]} + row = { + column: _cell(own[column] if column in own else getattr(run, column)) + for column in RUN_COLUMNS + } 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] @@ -773,7 +828,7 @@ def read_metric_names( @router.get("/{run_id}", response_model=RunDetail) -def read_run(run_id: str, session: SessionDep) -> Any: +def read_run(run_id: str, session: SessionDep, controller: FlowControllerDep) -> Any: """One run in full: what it was asked, what each node did, what it made.""" run = session.get(Run, run_id) if run is None: @@ -782,8 +837,16 @@ def read_run(run_id: str, session: SessionDep) -> Any: artifacts = session.exec( select(RunArtifact).where(col(RunArtifact.run_id) == run_id) ).all() - detail = RunDetail.model_validate(run, from_attributes=True) - detail.nodes = [RunNodeRow.model_validate(n, from_attributes=True) for n in nodes] + store = controller.store + detail = _named(RunDetail, [run], store)[0] + # Stored as `.`, shown as the engine names them. + shown = shown_nodes({n.node for n in nodes} | {a.node for a in artifacts}, store) + detail.nodes = [ + RunNodeRow.model_validate(n, from_attributes=True).model_copy( + update={"node": shown[n.node]} + ) + for n in nodes + ] # Worked out here, not stored: a source run is deleted long after the # nodes restored from it were written. sources = {n.cached_from for n in nodes if n.cached_from} @@ -795,19 +858,29 @@ def read_run(run_id: str, session: SessionDep) -> Any: for row in detail.nodes: row.curve_lost = bool(row.cached_from) and row.cached_from not in kept detail.artifacts = [ - ArtifactRow.model_validate(a, from_attributes=True) for a in artifacts + ArtifactRow.model_validate(a, from_attributes=True).model_copy( + update={"node": shown[a.node]} + ) + for a in artifacts ] return detail @router.get("/{run_id}/nodes/{node}/outputs", response_model=dict[str, Any] | None) -def read_node_outputs(run_id: str, node: str, session: SessionDep) -> Any: +def read_node_outputs( + run_id: str, node: str, session: SessionDep, controller: FlowControllerDep +) -> Any: """What one node returned, by message, as the stage cache stored it. - Kept only for a node that succeeded and may be reused, and only up to - 256 kB of JSON. Null when the node published nothing. + The node is ``.``, as the run shows it. Kept only for a node + that succeeded and may be reused, and only up to 256 kB of JSON. Null when + the node published nothing. """ - row = session.get(RunNode, {"run_id": run_id, "node": node}) + run = session.get(Run, run_id) + if run is None: + raise HTTPException(status_code=404, detail="No such run") + stored = stored_node(node, controller.store, run.flow_id) + row = session.get(RunNode, {"run_id": run_id, "node": stored}) if row is None: raise HTTPException(status_code=404, detail=f"No node '{node}' in this run") if row.outputs is None: @@ -823,7 +896,9 @@ def read_node_outputs(run_id: str, node: str, session: SessionDep) -> Any: @router.post("/{run_id}/cancel", response_model=RunRow) -async def cancel_run(run_id: str, request: Request, session: SessionDep) -> Any: +async def cancel_run( + run_id: str, request: Request, session: SessionDep, controller: FlowControllerDep +) -> Any: """Stop a run. One already past its last node is left as it finished.""" run = session.get(Run, run_id) if run is None: @@ -831,21 +906,24 @@ async def cancel_run(run_id: str, request: Request, session: SessionDep) -> Any: service = _service(request) await run_in_threadpool(service.cancel, run_id) session.refresh(run) - return run + return _named(RunRow, [run], controller.store)[0] @router.post("/{run_id}/retry", response_model=RunRow, status_code=202) -async def retry_run(run_id: str, request: Request, user: CurrentUser) -> Any: +async def retry_run( + run_id: str, request: Request, user: CurrentUser, controller: FlowControllerDep +) -> Any: """Run the same thing again, as a new run pointing back at this one.""" service = _service(request) try: - return await run_in_threadpool(service.retry, run_id, user.email) + run = await run_in_threadpool(service.retry, run_id, user.email) except FlowNotFound as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc except RunRejected as exc: detail = str(exc) status = 404 if detail.startswith("No run ") else 422 raise HTTPException(status_code=status, detail=detail) from exc + return _named(RunRow, [run], controller.store)[0] def _drop_runs(session: Session, chosen: Any) -> None: @@ -867,23 +945,38 @@ def _drop_runs(session: Session, chosen: Any) -> None: session.execute(delete(Run).where(col(Run.id).in_(chosen))) -def _live_run(session: Session, flow: str) -> str | None: +def _live_run(session: Session, flow_id: str) -> str | None: """A run of this flow the driver is still writing to, if there is one.""" return session.exec( select(Run.id).where( - col(Run.flow) == flow, col(Run.status).in_(("running", "queued")) + col(Run.flow_id) == flow_id, col(Run.status).in_(("running", "queued")) ) ).first() +def _audit(action: str, flow_id: str, user: CurrentUser, store: FlowStore) -> None: + """Record who deleted runs. On the bus a flow goes by its key and id.""" + event_bus.publish( + { + "type": "audit", + "action": action, + "flow": name_of(flow_id, store), + "flow_id": flow_id, + "user": user.email, + "ts": time.time(), + } + ) + + @router.delete("", response_model=Message) def delete_runs( session: SessionDep, user: CurrentUser, + controller: FlowControllerDep, flow: Annotated[str, Query(min_length=1)], history: bool = False, ) -> Any: - """Every run of one flow at once. + """Every run of one flow at once, ``flow`` being its id. The counterpart to the list's ``flow`` filter, and what a reseed needs: deleting a flow leaves its runs standing on purpose, so this is the only @@ -906,28 +999,22 @@ def delete_runs( ) # Counted before the delete, and a subquery rather than a materialised # list of ids on the way in: a demo can hold thousands. - chosen = select(col(Run.id)).where(col(Run.flow) == flow) + chosen = select(col(Run.id)).where(col(Run.flow_id) == flow) gone = session.exec( - select(func.count()).select_from(Run).where(col(Run.flow) == flow) + select(func.count()).select_from(Run).where(col(Run.flow_id) == flow) ).one() _drop_runs(session, chosen) if history: - session.execute(delete(FlowRun).where(col(FlowRun.flow) == flow)) + session.execute(delete(FlowRun).where(col(FlowRun.flow_id) == flow)) session.commit() - event_bus.publish( - { - "type": "audit", - "action": f"deleted {gone} run(s)", - "flow": flow, - "user": user.email, - "ts": time.time(), - } - ) + _audit(f"deleted {gone} run(s)", flow, user, controller.store) return Message(message=f"Deleted {gone} run(s) of '{flow}'") @router.delete("/{run_id}", status_code=204) -def delete_run(run_id: str, session: SessionDep, user: CurrentUser) -> Response: +def delete_run( + run_id: str, session: SessionDep, user: CurrentUser, controller: FlowControllerDep +) -> Response: """Forget a run and everything hanging off it. A live run is refused rather than raced: the driver writes its nodes back @@ -949,29 +1036,24 @@ def delete_run(run_id: str, session: SessionDep, user: CurrentUser) -> Response: "finish, before deleting it." ), ) - flow = run.flow + flow_id = run.flow_id _drop_runs(session, [run_id]) session.commit() - event_bus.publish( - { - "type": "audit", - "action": f"deleted run {run_id}", - "flow": flow, - "user": user.email, - "ts": time.time(), - } - ) + _audit(f"deleted run {run_id}", flow_id, user, controller.store) return Response(status_code=204) -def _series(session: Session, run_id: str, name: str = "") -> list[RunMetric]: +def _series( + session: Session, store: FlowStore, run_id: str, name: str = "" +) -> list[RunMetric]: """A run's numbers, including the ones a cached node points at. A cache hit replays no emissions, so a node restored from an earlier run has no rows of its own — it carries that run's id instead, and its series is read from there. Names are re-qualified on the way out, because the same node reached through two flows publishes under two names and the caller asked for - this run's. + this run's. Nodes are found by flow id, which is how rows name them, and + names are moved between the two flows' keys, which is how they are named. """ statement = select(RunMetric).where(col(RunMetric.run_id) == run_id) if name: @@ -985,16 +1067,17 @@ def _series(session: Session, run_id: str, name: str = "") -> list[RunMetric]: ).all() if restored: run = session.get(Run, run_id) - flow = run.flow if run is not None else "" + flow = run.flow_id if run is not None else "" # Two queries for the lot, rather than a `Run` lookup and a `RunMetric` # query per restored node. A comparison of twenty runs calls this # twenty times, so the per-node round trips multiplied. sources = { - row.id: row.flow + row.id: row.flow_id for row in session.exec( select(Run).where(col(Run.id).in_({n.cached_from for n in restored})) ) } + keys = names_of({flow, *sources.values()}, store) wanted: dict[tuple[str, str], list[RunNode]] = {} for node_row in restored: source_flow = sources.get(node_row.cached_from) @@ -1017,7 +1100,7 @@ def _series(session: Session, run_id: str, name: str = "") -> list[RunMetric]: for row in cached: for node_row in wanted.get((row.run_id, row.node), ()): source_flow = sources[node_row.cached_from] - renamed = requalify(row.name, source_flow, flow) + renamed = requalify(row.name, keys[source_flow], keys[flow]) if name and renamed != name: continue rows.append( @@ -1039,6 +1122,7 @@ def _series(session: Session, run_id: str, name: str = "") -> list[RunMetric]: def read_metrics( run_id: str, session: SessionDep, + controller: FlowControllerDep, name: str = "", stride: Annotated[int, Query(ge=1, le=100_000)] = 1, ) -> Any: @@ -1047,13 +1131,15 @@ def read_metrics( ``stride`` thins a long curve down: 3000 steps drawn on a 400-pixel chart is 3000 points nobody can see. """ - rows = _series(session, run_id, name) + rows = _series(session, controller.store, run_id, name) if stride > 1: rows = rows[:: max(1, stride)] return rows -def _points(session: Session, run_id: str, metric: str, x: str) -> list[list[float]]: +def _points( + session: Session, store: FlowStore, run_id: str, metric: str, x: str +) -> list[list[float]]: """One run's readings of ``metric``, against whichever x was asked for. The step is the default because it is what every run has. Time answers @@ -1063,20 +1149,26 @@ def _points(session: Session, run_id: str, metric: str, x: str) -> list[list[flo epoch, or samples seen — and is joined on the step the two share, which is the only thing they have in common. """ - rows = _series(session, run_id, metric) + rows = _series(session, store, run_id, metric) if x == "time": if not rows: return [] start = min(row.ts for row in rows) return [[row.ts - start, row.value] for row in rows] if x and x != "step": - against = {row.step: row.value for row in _series(session, run_id, x)} + against = {row.step: row.value for row in _series(session, store, run_id, x)} return [[against[row.step], row.value] for row in rows if row.step in against] return [[float(row.step), row.value] for row in rows] @router.get("/series/compare", response_model=SeriesAnswer) -def compare_metric(session: SessionDep, ids: str, metric: str, x: str = "") -> Any: +def compare_metric( + session: SessionDep, + controller: FlowControllerDep, + ids: str, + metric: str, + x: str = "", +) -> Any: """One metric across several runs, as the chart widget's series shape. This is the comparison view: it answers in the same shape a flow answers a @@ -1102,6 +1194,9 @@ def compare_metric(session: SessionDep, ids: str, metric: str, x: str = "") -> A if run.seed is not None: label = f"{run_id} (seed {run.seed})" lines.append( - MetricSeries(label=label, points=_points(session, run_id, metric, x)) + MetricSeries( + label=label, + points=_points(session, controller.store, run_id, metric, x), + ) ) return SeriesAnswer(metric=metric, x=x or "step", lines=lines) diff --git a/backend/fluksio/api/routes/search.py b/backend/fluksio/api/routes/search.py index e00b4c0..5ab51a6 100644 --- a/backend/fluksio/api/routes/search.py +++ b/backend/fluksio/api/routes/search.py @@ -54,16 +54,19 @@ class SearchEntry(BaseModel): """ category: Category - #: The id the frontend routes on. For an input or output, the qualified - #: message, which is what the canvas names its label by. For a setting, - #: ``.``: a node id has no dot, so the node is what comes - #: before the first one. + #: What it is called: a flow's or dashboard's key, or the id of anything + #: else. For an input or output, the qualified message, which is what the + #: canvas names its label by. For a setting, ``.``: a node + #: id has no dot, so the node is what comes before the first one. name: str #: Human title, often empty — a flow is usually only its name. title: str = "" #: The flow a node, input or output sits in, or the dashboard a widget - #: sits on. + #: sits on, by key. parent: str = "" + #: The id of the flow or dashboard opening it goes to: its own, or the one + #: it sits in. Empty for everything else, whose ``name`` is its id. + id: str = "" #: Node type, input dtype, widget type, channel kind. kind: str = "" #: What a setting is set to, as text. Only ever what ``setting_text`` lets @@ -159,13 +162,16 @@ def _build( flow = controller.store.read_flow(name, draft=True) except FlowNotFound: continue - entries.append(SearchEntry(category="flow", name=flow.name, title=flow.title)) + entries.append( + SearchEntry(category="flow", name=flow.name, title=flow.title, id=flow.id) + ) entries.extend( SearchEntry( category="node", name=node.id, title=node.title, parent=flow.name, + id=flow.id, kind=node.type, ) for node in flow.nodes @@ -184,6 +190,7 @@ def _build( category="setting", name=f"{node.id}.{key}", parent=flow.name, + id=flow.id, kind=node.type, detail=text, ) @@ -206,18 +213,22 @@ def _build( # The label the canvas draws: bare when it is the flow's own. title=message.removeprefix(f"{flow.name}."), parent=flow.name, + id=flow.id, kind=kind, ) ) for summary in dashboards.list(): try: - dashboard = dashboards.read(summary.name, draft=True) + dashboard = dashboards.read(summary.id, draft=True) except DashboardNotFound: continue entries.append( SearchEntry( - category="dashboard", name=dashboard.name, title=dashboard.title + category="dashboard", + name=dashboard.name, + title=dashboard.title, + id=dashboard.id, ) ) entries.extend( @@ -226,6 +237,7 @@ def _build( name=widget.id, title=widget.title, parent=dashboard.name, + id=dashboard.id, kind=widget.type, ) for widget in dashboard.widgets diff --git a/backend/fluksio/flow/controller.py b/backend/fluksio/flow/controller.py index 288a087..bf1018d 100644 --- a/backend/fluksio/flow/controller.py +++ b/backend/fluksio/flow/controller.py @@ -25,7 +25,7 @@ import traceback from collections import Counter from collections.abc import Callable from contextlib import suppress -from dataclasses import dataclass, field +from dataclasses import dataclass, field, replace from enum import Enum from typing import Any, TypeGuard, cast @@ -228,11 +228,13 @@ class RunContext: class Wiring: """One published node and the qualified messages it is bound to.""" + #: The flow's key, which the messages are qualified with, and its id. flow: str node: str type: str provides: list[str] requires: list[str] + flow_id: str = "" #: Nothing held, as distinct from a node that yielded ``None``. @@ -977,10 +979,9 @@ class FlowController: async def reload_lib_users(self, ref: str) -> list[str]: """Rebuild the flows running a shared node whose source has changed. - Returns their ids, which is what `usages` qualifies with. + Returns their ids. """ - usages = self.store.usages(ref, draft=False) - flows = sorted({usage.split(".", 1)[0] for usage in usages}) + flows = self.store.lib_users(ref) for flow in flows: await self.reload_flow(flow) return flows @@ -1718,19 +1719,20 @@ class FlowController: ) if provides or requires: index.append( - Wiring(other.name, node.id, node.type, provides, requires) + Wiring( + other.name, node.id, node.type, provides, requires, other.id + ) ) self._wiring_cache = (revision, index) return index - def cross_flow_nodes( - self, flow: str - ) -> list[tuple[str, str, list[str], list[str]]]: + def cross_flow_nodes(self, flow: str) -> list[Wiring]: """Nodes in other flows on the far side of a dotted message name. - Returns ``(flow, node id, messages of `flow` it provides, ones it - requires)``. This is what link-in and link-out mean: the wiring is - already real — the canvas simply had no way to draw the other end. + Returns the far node's `Wiring`, narrowed to the messages of `flow` it + provides and requires. This is what link-in and link-out mean: the + wiring is already real — the canvas simply had no way to draw the + other end. *flow* is a key. """ prefix = f"{flow}." found = [] @@ -1740,7 +1742,7 @@ class FlowController: provides = [name for name in entry.provides if name.startswith(prefix)] requires = [name for name in entry.requires if name.startswith(prefix)] if provides or requires: - found.append((entry.flow, entry.node, provides, requires)) + found.append(replace(entry, provides=provides, requires=requires)) return found def message_node( @@ -1818,6 +1820,7 @@ class FlowController: group.members.append(member) if flow.name not in group.flows: group.flows.append(flow.name) + group.flow_ids.append(flow.id) # A merged neuron stands for several nodes, so one of them being # unable to run is enough to mark it; the first reason wins. if group.issue is None: diff --git a/backend/fluksio/flow/dashboards.py b/backend/fluksio/flow/dashboards.py index 7b7521d..8a9b252 100644 --- a/backend/fluksio/flow/dashboards.py +++ b/backend/fluksio/flow/dashboards.py @@ -979,7 +979,7 @@ class DashboardStore: return copy def bindings_for(self, flow: str) -> Bindings: - """Every widget bound to a message of ``flow``. + """Every widget bound to a message of the flow with key ``flow``. What the canvas draws as an endpoint: a control that sets one of this flow's messages, or a tile that shows one. Without this a dashboard is @@ -993,6 +993,8 @@ class DashboardStore: defn = DashboardDef.model_validate_json(path.read_text()) except Exception: continue + # By id, which is what opening it takes; the title is the label. + dash_id = defn.id or path.parent.name # A bound setting is a consumer too — the dashboard itself reading # a message rather than any tile on it — so the canvas accounts for # it the same way. ``widget`` is what the endpoint id is built @@ -1003,7 +1005,7 @@ class DashboardStore: continue found.append( { - "dashboard": defn.name, + "dashboard": dash_id, "dashboard_title": defn.title or defn.name, "widget": f"settings.{name}", "title": f"{defn.title or defn.name} {name}", @@ -1020,7 +1022,7 @@ class DashboardStore: continue found.append( { - "dashboard": defn.name, + "dashboard": dash_id, "dashboard_title": defn.title or defn.name, "widget": widget.id, "title": widget.title or widget.id, diff --git a/backend/fluksio/flow/devices.py b/backend/fluksio/flow/devices.py index 3012406..5be8688 100644 --- a/backend/fluksio/flow/devices.py +++ b/backend/fluksio/flow/devices.py @@ -77,6 +77,7 @@ class Binding: """One port of a device, wired to a message by a device node.""" device: str + #: The flow's key, which ``message`` is qualified with. flow: str node: str port: str @@ -88,6 +89,8 @@ class Binding: dtype: str #: Passed through to the device unread. config: dict[str, Any] = field(default_factory=dict) + #: The flow's id, which is where wiring the port is edited. + flow_id: str = "" class DeviceIo(BaseModel): @@ -102,7 +105,9 @@ class DeviceIo(BaseModel): class DeviceBinding(BaseModel): """Where a port is wired: the device node holding it.""" + #: The flow's key, and its id to open it by. flow: str + flow_id: str = "" node: str port: str message: str @@ -129,6 +134,7 @@ def bindings_of(flow: FlowDef) -> list[Binding]: Binding( device=device, flow=flow.name, + flow_id=flow.id, node=node.id, port=spec.port, message=qualify(flow.name, spec.name), @@ -145,9 +151,10 @@ class DeviceDef(BaseModel): id: str title: str = "" - #: Ordered. The first one is what the device opens after pairing, and the - #: rail follows this order. A name that no longer resolves is a dashboard - #: someone deleted, and the device skips it. + #: Dashboard ids, ordered. The first one is what the device opens after + #: pairing, and the rail follows this order. An id that no longer resolves + #: is a dashboard someone deleted, and the device skips it. A dashboard from + #: before ids has its key as its id, so a list from then still holds. dashboards: list[str] = Field(default_factory=list) #: Bigger controls, a bigger rail and no hover states, for a screen that is #: touched rather than pointed at. A fact about the screen, not about any @@ -513,6 +520,7 @@ class DeviceStore: ( DeviceBinding( flow=b.flow, + flow_id=b.flow_id, node=b.node, port=b.port, message=b.message, @@ -1013,7 +1021,7 @@ def dashboards_for( ) -> list[DashboardDef]: """The published documents this device shows, in rail order. - Published, since that is what a device draws. A name that no longer + Published, since that is what a device draws. An id that no longer resolves is skipped, and a device that is gone shows nothing. Handed back whole rather than walked here, because what a device may do @@ -1022,7 +1030,7 @@ def dashboards_for( """ if device is None: return [] - return [defn for name in device.dashboards if (defn := store.published(name))] + return [defn for dash_id in device.dashboards if (defn := store.published(dash_id))] def messages_of(defn: DashboardDef) -> set[str]: diff --git a/backend/fluksio/flow/events.py b/backend/fluksio/flow/events.py index 314caec..337d4e0 100644 --- a/backend/fluksio/flow/events.py +++ b/backend/fluksio/flow/events.py @@ -13,7 +13,7 @@ import asyncio import logging import threading from collections import deque -from collections.abc import AsyncIterator +from collections.abc import AsyncIterator, Callable from contextlib import asynccontextmanager from typing import Any @@ -118,6 +118,11 @@ class EventBus: #: Subscribers that lost an event since they last asked. Touched on #: the event loop only, by the flush and the subscribers. self._dropped: set[asyncio.Queue[dict[str, Any]]] = set() + #: The id of the flow holding a key, once the store exists. An event + #: naming a flow by its key gets its ``flow_id`` here, on the way in, + #: so what holds the event later is not left to look up a key that + #: may have moved on by then. + self.flow_id_of: Callable[[str], str] | None = None def bind(self, loop: asyncio.AbstractEventLoop) -> None: """Attach the bus to the running event loop (called once at startup).""" @@ -125,6 +130,11 @@ class EventBus: def publish(self, event: dict[str, Any]) -> None: """Publish an event from any thread.""" + resolve = self.flow_id_of + if resolve is not None and "flow" in event and "flow_id" not in event: + # An event about the engine as a whole names no flow, and no id. + flow = str(event["flow"] or "") + event["flow_id"] = resolve(flow) if flow else "" kind = event.get("type") if kind == "node_log": # deque.append is atomic, so worker threads need no lock here. diff --git a/backend/fluksio/flow/metrics.py b/backend/fluksio/flow/metrics.py index 719c54c..08b7312 100644 --- a/backend/fluksio/flow/metrics.py +++ b/backend/fluksio/flow/metrics.py @@ -26,6 +26,7 @@ from fluksio.core.config import settings from fluksio.core.db import engine from fluksio.flow import tombstones from fluksio.flow.events import EventBus +from fluksio.flow.messages import requalify from fluksio.models import EngineEvent, FlowRun, MetricBucket logger = logging.getLogger(__name__) @@ -84,6 +85,19 @@ def _minute(ts: float) -> int: return int(ts // 60) * 60 +def _filed(event: dict[str, Any], node: str = "") -> tuple[str, str]: + """The flow id and the ``.`` an event is kept under. + + The event names both by the flow's key and brings the id along from the + bus. One from a bus that knows no ids is kept under the key, which is + what the id of a flow from before ids is. + """ + flow = str(event.get("flow") or "") + flow_id = str(event.get("flow_id") or flow) + node = node or str(event.get("node") or "") + return flow_id[:NAME_MAX], requalify(node, flow, flow_id)[:NAME_MAX] + + def _detail(event: dict[str, Any]) -> str: text = str(event.get("error") or event.get("reason") or event.get("detail") or "") if event.get("type") == "cascade_dropped": @@ -154,11 +168,7 @@ class MetricsCollector: def _bucket(self, event: dict[str, Any]) -> dict[str, float]: # ponytail: one collector, one row per node per minute; coarsen the # bucket if the node count ever reaches thousands. - key = ( - str(event.get("flow") or "")[:NAME_MAX], - str(event.get("node") or "")[:NAME_MAX], - _minute(float(event.get("ts") or time.time())), - ) + key = (*_filed(event), _minute(float(event.get("ts") or time.time()))) return self._buckets.setdefault( key, { @@ -216,12 +226,13 @@ class MetricsCollector: key = (str(event.get("flow") or ""), str(event.get("node") or "")) traceback = self._tracebacks.pop(key, (0.0, ""))[1] error = str(event.get("error") or "") + flow_id, node = _filed(event) self._pending.append( EngineEvent( ts=datetime.fromtimestamp(ts, UTC), type="node_error", - flow=str(event.get("flow") or ""), - node=str(event.get("node") or ""), + flow_id=flow_id, + node=node, detail=(f"{error}\n{traceback}" if traceback else error)[ :DETAIL_CAP ], @@ -252,12 +263,13 @@ class MetricsCollector: if kind == "node_health": if event.get("health") == "down": + flow_id, node = _filed(event) self._pending.append( EngineEvent( ts=datetime.fromtimestamp(ts, UTC), type="node_health", - flow=str(event.get("flow") or ""), - node=str(event.get("node") or ""), + flow_id=flow_id, + node=node, detail=_detail(event) or "Reported itself down.", run=run_id, ) @@ -269,7 +281,7 @@ class MetricsCollector: EngineEvent( ts=datetime.fromtimestamp(ts, UTC), type="audit", - flow=str(event.get("flow") or ""), + flow_id=_filed(event)[0], detail=str(event.get("action") or ""), actor=str(event.get("user") or ""), ) @@ -277,12 +289,13 @@ class MetricsCollector: return if kind in RECORDED: + flow_id, node = _filed(event, str(event.get("task") or "")) self._pending.append( EngineEvent( ts=datetime.fromtimestamp(ts, UTC), type=kind, - flow=str(event.get("flow") or ""), - node=str(event.get("node") or event.get("task") or ""), + flow_id=flow_id, + node=node, detail=_detail(event), run=run_id, ) @@ -307,7 +320,7 @@ class MetricsCollector: return self._runs[run_id] = { "id": run_id, - "flow": str(event.get("flow") or "")[:NAME_MAX], + "flow_id": _filed(event)[0], "source": str(event.get("cause") or ""), "parent": str(event.get("parent") or "")[:64], "started_ts": ts, @@ -417,12 +430,12 @@ class MetricsCollector: ), [ { - "flow": flow, + "flow_id": flow_id, "node": node, "bucket": datetime.fromtimestamp(minute, UTC), **agg, } - for (flow, node, minute), agg in buckets.items() + for (flow_id, node, minute), agg in buckets.items() ], ) @@ -434,10 +447,11 @@ class MetricsCollector: session.execute( statement.on_conflict_do_update( index_elements=["id"], + # Not the flow: a run is of the one it started in. set_={ key: statement.excluded[key] for key in values[0] - if key != "id" + if key not in ("id", "flow_id") }, ), values, diff --git a/backend/fluksio/flow/runs.py b/backend/fluksio/flow/runs.py index 5b98a04..e7d1728 100644 --- a/backend/fluksio/flow/runs.py +++ b/backend/fluksio/flow/runs.py @@ -49,9 +49,10 @@ from sqlalchemy.exc import IntegrityError from sqlmodel import Session, col, select from fluksio.core.db import engine as db_engine +from fluksio.flow import tombstones from fluksio.flow.artifacts import ArtifactStore, is_reference, valid_digest from fluksio.flow.controller import NODE_TYPES, FlowController, RunContext -from fluksio.flow.messages import qualify +from fluksio.flow.messages import qualify, requalify from fluksio.flow.pipeline import CACHE_SHAPE_SINCE, CacheHit, NodeOutcome, Pipeline from fluksio.flow.queue import WorkItem, WorkQueue from fluksio.flow.schemas import FlowDef, NodeDef @@ -531,8 +532,14 @@ class MetricSink: batch: int = METRIC_BATCH, interval: float = METRIC_FLUSH_S, on_flush: Callable[[list[str]], None] | None = None, + key: str = "", + flow_id: str = "", ) -> None: self.run_id = run_id + #: The run's flow, by the key its nodes are named with and by the id + #: a row names them with: ``.``. + self._key = key + self._flow_id = flow_id self._batch = batch self._interval = interval self._on_flush = on_flush @@ -562,7 +569,7 @@ class MetricSink: run_id=self.run_id, name=name[:128], step=step, - node=node_id[:255], + node=requalify(node_id, self._key, self._flow_id)[:255], ts=now, value=float(value), ) @@ -802,8 +809,17 @@ class RunCache: not a hit: the reference would name bytes nobody can open. """ - def __init__(self, artifacts: ArtifactStore | None = None) -> None: + def __init__( + self, + artifacts: ArtifactStore | None = None, + key_of: Callable[[str], str] | None = None, + ) -> None: self.artifacts = artifacts + #: The key a source run's outputs are named with, from its flow's id: + #: the flow's key now, or the last one it had if it is gone. Without + #: one the id is taken for the key, which it is for a flow from before + #: ids. + self.key_of = key_of or (lambda flow_id: flow_id) def lookup(self, key: str) -> CacheHit | None: if not key: @@ -829,7 +845,7 @@ class RunCache: continue source = session.get(Run, row.run_id) return CacheHit( - flow=source.flow if source is not None else "", + flow=self.key_of(source.flow_id) if source else "", outputs=outputs, # A row that was itself restored holds no series, so # follow its pointer rather than adding a hop to it. @@ -862,7 +878,9 @@ class RunService: self.controller = controller self.queue = queue self._artifacts = artifacts - self._cache = RunCache(artifacts) + self._cache = RunCache( + artifacts, lambda flow_id: tombstones.name_of(flow_id, controller.store) + ) # Without one, a run gets a private in-memory state — which is exactly # the isolation it wants, minus surviving the process. self._state_factory = state_factory or (lambda _ns: MemoryState()) @@ -912,10 +930,11 @@ class RunService: """ with Session(db_engine) as session: waiting = session.exec( - select(Run.id, Run.flow).where(col(Run.status) == "queued") + select(Run.id, Run.flow_id).where(col(Run.status) == "queued") ).all() - for run_id, flow in waiting: - self.queue.add(WorkItem(kind="run", node="", flow=flow, run_id=run_id)) + # A run item names its flow by id, as `submit` queues it. + for run_id, flow_id in waiting: + self.queue.add(WorkItem(kind="run", node="", flow=flow_id, run_id=run_id)) if waiting: logger.info( "Requeued %d run(s) left queued by an earlier engine", len(waiting) @@ -940,7 +959,7 @@ class RunService: def submit( self, - flow_name: str, + flow_id: str, params: dict[str, Any] | None = None, seed: int | None = None, group_id: str | None = None, @@ -953,10 +972,12 @@ class RunService: passed: list[str] | None = None, for_node: str | None = None, ) -> Run: - """Journal a run and wake an engine up for it. Never blocks on it. + """Journal a run of the flow *flow_id* and wake an engine up for it. + Never blocks on it. ``passed`` overrides which of ``params`` count as the caller's own, for - a retry handing on its parent's whole params. + a retry handing on its parent's whole params. ``for_node`` is a + ``.``. """ # Before anything else, including reading the flow: a caller retrying a # submit it never got an answer for is owed the run that answer was @@ -965,7 +986,7 @@ class RunService: existing = self._by_key(idempotency_key) if existing is not None: return existing - flow = self.controller.store.read_flow(flow_name, draft=draft) + flow = self.controller.store.read_flow(flow_id, draft=draft) issues = batch_issues(flow, self.controller.store, draft=draft) if issues: raise RunRejected(" ".join(issues)) @@ -973,7 +994,7 @@ class RunService: run = Run( id=new_run_id(), - flow=flow.name, + flow_id=flow.id, flow_version=flow.version, commit=self.controller.store.head(), origin_commit=origin_commit(flow), @@ -1011,7 +1032,8 @@ class RunService: return existing session.refresh(run) - self.queue.add(WorkItem(kind="run", node="", flow=flow.name, run_id=run.id)) + # By id, like the run: what the item names outlives a rename of the key. + self.queue.add(WorkItem(kind="run", node="", flow=flow.id, run_id=run.id)) return run def _complete( @@ -1087,10 +1109,10 @@ class RunService: {"external_id": external_id, "id": existing, "created": False} ) continue - name = str(entry["flow"]) - if name not in flows: - flows[name] = self.controller.store.read_flow(name) - flow = flows[name] + flow_id = str(entry["flow"]) + if flow_id not in flows: + flows[flow_id] = self.controller.store.read_flow(flow_id) + flow = flows[flow_id] try: params, seed, passed = self._complete( flow, entry["params"], entry["seed"] @@ -1100,7 +1122,7 @@ class RunService: started, finished = entry["started_at"], entry["finished_at"] run = Run( id=new_run_id(), - flow=flow.name, + flow_id=flow.id, flow_version=flow.version, params=params, passed=passed, @@ -1207,7 +1229,7 @@ class RunService: # The gate first, so nothing new is submitted while the running nodes # are being killed; a gated node is never handed to the executor, so # the graph drains instead of going further. - pipeline.pause(self._flow_of(run_id) or "") + pipeline.pause(self._key_of(run_id)) workers = self.controller.workers if workers is not None: # Keyed by run, so a sweep cancelling one config leaves the others @@ -1239,7 +1261,7 @@ class RunService: raise RunRejected("That run has not finished; cancel it first") source = run.model_copy() return self.submit( - source.flow, + source.flow_id, params=dict(source.params), seed=source.seed, group_id=source.group_id, @@ -1254,10 +1276,11 @@ class RunService: for_node=source.for_node, ) - def _flow_of(self, run_id: str) -> str | None: + def _key_of(self, run_id: str) -> str: + """The key a run's pipeline runs its flow under, which it pauses by.""" with Session(db_engine) as session: run = session.get(Run, run_id) - return run.flow if run else None + return tombstones.name_of(run.flow_id, self.controller.store) if run else "" def _finish_unstarted(self, run_id: str, status: str) -> bool: """Cancel a run nothing of has executed, if it is still in ``status``.""" @@ -1464,23 +1487,31 @@ class RunService: state: StateBackend | None = None errors = 0 + # The key its nodes and messages are named with, the id its rows are + # filed under. Read from the store and its tombstones rather than the + # flow, which may be gone before the run gets as far as reading it. + key = tombstones.name_of(run.flow_id, self.controller.store) + def observe(outcome: NodeOutcome) -> None: nonlocal errors if not outcome.ok: errors += 1 - self._record_node(run_id, outcome) + self._record_node(run, key, outcome) sink = MetricSink( run_id, on_flush=lambda names: self._publish_event( { "type": "run_metric", - "flow": run.flow, + "flow": key, + "flow_id": run.flow_id, "run": run_id, "names": names, "ts": time.time(), } ), + key=key, + flow_id=run.flow_id, ) try: with self._lock: @@ -1488,7 +1519,7 @@ class RunService: # building one loads every python node in a worker, and waits # for one while all of them are busy — for longer than a lease. self._active[run_id] = None - flow = self.controller.store.read_flow(run.flow, draft=run.draft) + flow = self.controller.store.read_flow(run.flow_id, draft=run.draft) # Read again, now that it is this run's turn: a sweep queues every # run at once, and the code on disk is free to move in the hours # before the last of them starts. What the record must name is the @@ -1518,7 +1549,7 @@ class RunService: status, reason = "cancelled", "Cancelled before it started" return self._active[run_id] = pipeline - self._publish(run, "run_started") + self._publish(run, key, "run_started") pipeline.run(seed_values(flow, run.params, run.seed)) result = collect_result(flow, state) with self._lock: @@ -1542,7 +1573,7 @@ class RunService: duration = round((time.perf_counter() - started) * 1000, 2) self._finish(run_id, status, reason, result, duration) run.status = status - self._publish(run, "run_finished") + self._publish(run, key, "run_finished") self._release_cards(run) # Its values were only ever this run's; nothing reads them once it # has a result. On Redis the namespace would expire anyway. @@ -1582,11 +1613,15 @@ class RunService: except Exception: logger.warning("Could not retire the GPU workers of run %s", run.id) - def _record_node(self, run_id: str, outcome: NodeOutcome) -> None: + def _record_node(self, run: Run, key: str, outcome: NodeOutcome) -> None: + """Write what one node did, under ``.``. The outputs + and artifacts stay named by message, which is by the flow's *key*.""" + run_id = run.id + node = requalify(outcome.node, key, run.flow_id)[:255] outputs = _cacheable(outcome) row = RunNode( run_id=run_id, - node=outcome.node[:255], + node=node, status="cached" if outcome.cached else ("ok" if outcome.ok else "error"), started_at=datetime.now(UTC), duration_ms=outcome.duration_ms, @@ -1614,7 +1649,7 @@ class RunService: run_id=run_id, name=message[:255], filename=str(ref.get("name") or "")[:255] or None, - node=outcome.node[:255], + node=node, digest=str(ref.get("digest") or "")[:71], size=int(ref.get("size") or 0), media_type=str( @@ -1667,11 +1702,12 @@ class RunService: except Exception: logger.exception("Could not close run %s", run_id) - def _publish(self, run: Run, kind: str) -> None: + def _publish(self, run: Run, key: str, kind: str) -> None: self._publish_event( { "type": kind, - "flow": run.flow, + "flow": key, + "flow_id": run.flow_id, "run": run.id, "status": run.status, "group": run.group_id or "", diff --git a/backend/fluksio/flow/schemas.py b/backend/fluksio/flow/schemas.py index 63ba469..d5a94b7 100644 --- a/backend/fluksio/flow/schemas.py +++ b/backend/fluksio/flow/schemas.py @@ -529,7 +529,10 @@ class BrainNode(BaseModel): label: str kind: str members: list[str] = Field(default_factory=list) + #: The keys of the flows its members sit in, and their ids in the same + #: order, which is what opening one goes by. flows: list[str] = Field(default_factory=list) + flow_ids: list[str] = Field(default_factory=list) issue: str | None = Field( default=None, description=( diff --git a/backend/fluksio/flow/store.py b/backend/fluksio/flow/store.py index 3e0f445..3374ec9 100644 --- a/backend/fluksio/flow/store.py +++ b/backend/fluksio/flow/store.py @@ -29,12 +29,12 @@ import shutil import string import subprocess import threading -from collections.abc import Callable, Iterable +from collections.abc import Callable, Iterable, Iterator from datetime import UTC, datetime from pathlib import Path from typing import TYPE_CHECKING, Literal -from fluksio.flow.schemas import FlowDef, _validate_name +from fluksio.flow.schemas import FlowDef, NodeDef, _validate_name if TYPE_CHECKING: from fluksio.flow.tombstones import Tombstones @@ -603,23 +603,30 @@ class FlowStore: self._commit(f"Detach and delete shared node '{name}'") def usages(self, lib_name: str, draft: bool = True) -> list[str]: - """Every ``flow.node`` pointing at this shared source. + """Every ``.`` pointing at this shared source. Read from the drafts the editor shows, or with ``draft`` off from the published flows the engine runs. """ - used = [] - for name in self.list_flows(): + return sorted( + f"{flow.name}.{node.id}" for flow, node in self._lib_nodes(lib_name, draft) + ) + + def lib_users(self, lib_name: str) -> list[str]: + """The ids of the published flows running this shared source.""" + return sorted({flow.id for flow, _ in self._lib_nodes(lib_name, False)}) + + def _lib_nodes( + self, lib_name: str, draft: bool + ) -> Iterator[tuple[FlowDef, NodeDef]]: + for flow_id in self.list_flows(): try: - definition = self.read_flow(name, draft=draft) + definition = self.read_flow(flow_id, draft=draft) except FlowNotFound: continue - used += [ - f"{name}.{node.id}" - for node in definition.nodes - if node.source_ref == lib_name - ] - return sorted(used) + for node in definition.nodes: + if node.source_ref == lib_name: + yield definition, node def _set_source_ref(self, flow: str, node_id: str, ref: str | None) -> None: """Point a node at a shared source, in the draft the editor is on.""" diff --git a/backend/fluksio/flow/tombstones.py b/backend/fluksio/flow/tombstones.py index b40e7db..19b3e91 100644 --- a/backend/fluksio/flow/tombstones.py +++ b/backend/fluksio/flow/tombstones.py @@ -5,6 +5,10 @@ flow's id is opaque, so a tombstone keeps the key it last had, and `names_of` reads that back for anything showing the history. A flow deleted before tombstones existed has its old key as its id, which already reads right. +History names a node ``.``. The API shows and takes it as +``.``, the way the engine names it, and `shown_nodes` and +`stored_node` translate between the two. + The flow store is the one writer: it buries a flow before removing its files, asks before issuing an id, and lifts the tombstone of a flow restored under a pinned id once the restored document is on disk. @@ -35,20 +39,20 @@ def referenced(session: Session, flow_id: str) -> bool: queries = [ select(Run.id).where( or_( - col(Run.flow) == flow_id, + col(Run.flow_id) == flow_id, col(Run.for_node).startswith(nodes, autoescape=True), ) ), - select(FlowRun.id).where(col(FlowRun.flow) == flow_id), - select(MetricBucket.flow).where( + select(FlowRun.id).where(col(FlowRun.flow_id) == flow_id), + select(MetricBucket.flow_id).where( or_( - col(MetricBucket.flow) == flow_id, + col(MetricBucket.flow_id) == flow_id, col(MetricBucket.node).startswith(nodes, autoescape=True), ) ), select(EngineEvent.id).where( or_( - col(EngineEvent.flow) == flow_id, + col(EngineEvent.flow_id) == flow_id, col(EngineEvent.node).startswith(nodes, autoescape=True), ) ), @@ -118,6 +122,44 @@ def name_of(flow_id: str, store: FlowStore) -> str: return names_of([flow_id], store)[flow_id] +def id_of(key: str, store: FlowStore) -> str: + """The flow id behind *key*: the live flow's, else that of the one deleted + flow that last held it, else *key* itself, as for history from before ids. + """ + live = store.key_index().get(key) + if live is not None: + return live + with Session(engine) as session: + graves = session.exec( + select(FlowTombstone.id).where(col(FlowTombstone.name) == key).limit(2) + ).all() + return graves[0] if len(graves) == 1 else key + + +def shown_nodes(nodes: Iterable[str], store: FlowStore) -> dict[str, str]: + """Each stored ``.`` as ``.``, with one lookup.""" + split = {node: node.partition(".") for node in nodes} + names = names_of({flow for flow, dot, _ in split.values() if dot}, store) + return { + node: f"{names[flow]}.{rest}" if dot else node + for node, (flow, dot, rest) in split.items() + } + + +def stored_node(node: str, store: FlowStore, flow_id: str = "") -> str: + """A ``.`` as history stores it, ``.``. + + With *flow_id*, a key that is that flow's own resolves to it, which is how + a node of a deleted or renamed flow is still found by the name it shows. + """ + key, dot, rest = node.partition(".") + if not dot: + return node + if flow_id and name_of(flow_id, store) == key: + return f"{flow_id}.{rest}" + return f"{id_of(key, store)}.{rest}" + + def prune(session: Session) -> int: """Drop the tombstones no history row names any more. Returns how many.""" stale = [ diff --git a/backend/fluksio/main.py b/backend/fluksio/main.py index febc112..95e4af0 100644 --- a/backend/fluksio/main.py +++ b/backend/fluksio/main.py @@ -308,6 +308,10 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: # after them write documents back with whatever id they read. await run_in_threadpool(assign_ids, settings.FLOWS_DIR) store = FlowStore(settings.FLOWS_DIR, tombstones=Tombstones()) + # Events name a flow by its key, and carry its id from here on. A key + # no flow holds any more is passed on as it is, which is the id of a + # flow from before ids. + event_bus.flow_id_of = lambda key: store.key_index().get(key, key) # The packages node code imports. In the background, because a first # install of anything substantial takes minutes and `uv` is given five # of them twice over — held here, the container never answers its diff --git a/backend/fluksio/models.py b/backend/fluksio/models.py index 402b26d..61b85e3 100644 --- a/backend/fluksio/models.py +++ b/backend/fluksio/models.py @@ -3,7 +3,7 @@ from datetime import UTC, datetime from typing import Any from pydantic import EmailStr -from sqlalchemy import JSON, Column, Index +from sqlalchemy import JSON, Column, Index, String from sqlmodel import Field, SQLModel from fluksio.core.types import UTCDateTime @@ -237,7 +237,8 @@ class MetricBucket(SQLModel, table=True): __tablename__ = "metric_minute" - flow: str = Field(primary_key=True, max_length=255) + flow_id: str = Field(sa_column=Column("flow", String(255), primary_key=True)) + #: ``.``. node: str = Field(primary_key=True, max_length=255) bucket: datetime = Field( primary_key=True, @@ -273,7 +274,9 @@ class EngineEvent(SQLModel, table=True): ) #: node_error, flow_quarantined, engine_degraded, …, or audit. type: str = Field(max_length=32, index=True) - flow: str = "" + #: The flow's id, empty for the engine as a whole. + flow_id: str = Field(default="", sa_column=Column("flow", String, nullable=False)) + #: ``.``, or empty. node: str = "" #: The error and its traceback, or what an audit entry says was done. detail: str = "" @@ -298,7 +301,9 @@ class FlowRun(SQLModel, table=True): #: The queue entry id, a `manual-` one for a run that never queued, or the #: `Run.id` of a batch run. id: str = Field(primary_key=True, max_length=64) - flow: str = Field(index=True, max_length=255) + flow_id: str = Field( + sa_column=Column("flow", String(255), index=True, nullable=False) + ) #: What caused it: the work item's cause, "manual", or "run" for a batch. source: str = "" #: The run this one continues: on a rate-limit flush, the run whose value @@ -345,7 +350,9 @@ class Run(SQLModel, table=True): ) id: str = Field(primary_key=True, max_length=64) - flow: str = Field(index=True, max_length=255) + flow_id: str = Field( + sa_column=Column("flow", String(255), index=True, nullable=False) + ) #: The flow document this ran, and the commit it was read at, so a result #: can be traced back to the code that produced it. flow_version: int = 1 @@ -390,8 +397,9 @@ class Run(SQLModel, table=True): ) #: Where it was asked for: api, cli, sdk, hook, sweep, import or train. cause: str = Field(default="api", max_length=32) - #: The node it was asked for from, as `flow.node`: the wren node a - #: training run is for. Null for every other run. + #: The node it was asked for from, as ``.``: the wren node a + #: training run is for, in another flow than the run's own. Null for every + #: other run. for_node: str | None = Field(default=None, max_length=255) #: Re-execute every node, whatever the stage cache holds for it. no_cache: bool = False diff --git a/backend/tests/api/routes/test_devices.py b/backend/tests/api/routes/test_devices.py index daae687..aa83a69 100644 --- a/backend/tests/api/routes/test_devices.py +++ b/backend/tests/api/routes/test_devices.py @@ -629,9 +629,12 @@ def test_a_devices_io_comes_from_its_device_nodes( "config": {"pull": "up", "debounce_ms": 50}, }, ] - assert [(b["flow"], b["node"], b["port"]) for b in read["bindings"]] == [ - ("porch", "porch", "backlight"), - ("porch", "porch", "gpio.4"), + # The flow by key, and by the id that opens it. + assert [ + (b["flow"], b["flow_id"], b["node"], b["port"]) for b in read["bindings"] + ] == [ + ("porch", "porch", "porch", "backlight"), + ("porch", "porch", "porch", "gpio.4"), ] again = client.get(f"{PANELS}/porch_io", headers=superuser_token_headers) assert again.content == first.content diff --git a/backend/tests/api/routes/test_flow_ids.py b/backend/tests/api/routes/test_flow_ids.py index 5f0fb70..651eb5c 100644 --- a/backend/tests/api/routes/test_flow_ids.py +++ b/backend/tests/api/routes/test_flow_ids.py @@ -16,6 +16,7 @@ from fastapi.testclient import TestClient from sqlmodel import Session, delete, select from fluksio.core.config import settings +from fluksio.flow.dashboards import DashboardDef, WidgetDef from fluksio.flow.store import ISSUED_ID from fluksio.flow.tombstones import names_of from fluksio.models import FlowTombstone, Run @@ -235,6 +236,65 @@ def test_a_results_dashboard_is_made_from_a_flow_id_under_the_flows_key( client.delete(f"{DASHBOARDS}/{board['id']}", headers=superuser_token_headers) +def test_what_opens_a_flow_or_dashboard_names_it_by_id( + make: Make, client: TestClient, superuser_token_headers: dict[str, str] +) -> None: + """A node goes by ``.``, which is what the canvas matches. Where + a response leads somewhere, it says the id to go by beside it.""" + headers = superuser_token_headers + key, reader_key = a_key(), a_key() + flow_id = make(key)["definition"]["id"] + for node, code in (("sensor", SENSOR), ("double", DOUBLE)): + client.put( + f"{FLOWS}/{flow_id}/nodes/{node}/source", + headers=headers, + json={"code": code}, + ) + client.post( + f"{FLOWS}/{flow_id}/publish", + headers=headers, + json={ + "version": client.get(f"{FLOWS}/{flow_id}", headers=headers).json()[ + "definition" + ]["version"] + }, + ) + listen = { + "id": "listen", + "type": "python", + "requires": [{"name": f"{key}.reading", "dtype": "float"}], + } + reader_id = make(reader_key, nodes=[listen])["definition"]["id"] + board = client.app.state.dashboard_store.write( # type: ignore[attr-defined] + DashboardDef( + id=f"d_{reader_key}", + name=reader_key, + widgets=[ + WidgetDef( + id="shown", + type="stat", + config={"message": f"{key}.reading", "dtype": "float"}, + ) + ], + ) + ) + try: + far = client.get(f"{FLOWS}/{reader_id}", headers=headers).json()["endpoints"] + assert [(e["id"], e["owner_id"]) for e in far] == [ + (f"flow:{key}.sensor", flow_id) + ] + near = client.get(f"{FLOWS}/{flow_id}", headers=headers).json()["endpoints"] + assert (f"dashboard:{board.id}:shown", board.id) in [ + (e["id"], e["owner_id"]) for e in near + ] + + graph = client.get(f"{FLOWS}/graph", headers=headers).json() + (sensor,) = [n for n in graph["nodes"] if f"{key}.sensor" in n["members"]] + assert (sensor["flows"], sensor["flow_ids"]) == ([key], [flow_id]) + finally: + client.delete(f"{DASHBOARDS}/{board.id}", headers=headers) + + def test_who_may_create_what( client: TestClient, superuser_token_headers: dict[str, str], @@ -286,7 +346,7 @@ def test_a_deleted_flow_keeps_its_name_and_frees_its_key( ) -> None: key = a_key() old = make(key)["definition"]["id"] - db.add(Run(id=f"run-{key}", flow=old, status="ok", created_at=datetime.now(UTC))) + db.add(Run(id=f"run-{key}", flow_id=old, status="ok", created_at=datetime.now(UTC))) db.commit() controller = client.app.state.flow_controller # type: ignore[attr-defined] controller.state.set(f"{key}.doubled", 1.0) @@ -311,7 +371,7 @@ def test_a_deleted_flow_keeps_its_name_and_frees_its_key( assert made.status_code == 201, made.text new = made.json()["definition"]["id"] assert new != old - assert db.exec(select(Run).where(Run.flow == new)).first() is None + assert db.exec(select(Run).where(Run.flow_id == new)).first() is None client.delete(f"{FLOWS}/{new}", headers=superuser_token_headers) db.exec(delete(Run).where(Run.id == f"run-{key}")) # type: ignore[call-overload] db.commit() diff --git a/backend/tests/api/routes/test_flows.py b/backend/tests/api/routes/test_flows.py index 25ee7a5..9256ad7 100644 --- a/backend/tests/api/routes/test_flows.py +++ b/backend/tests/api/routes/test_flows.py @@ -303,7 +303,7 @@ def test_delete_flow( put_flow(client, superuser_token_headers, a_flow(), "demo") # Finished: a flow with a live run refuses to be deleted, which is its own # test below. - db.add(Run(id="run-1", flow="demo", status="ok", created_at=datetime.now(UTC))) + db.add(Run(id="run-1", flow_id="demo", status="ok", created_at=datetime.now(UTC))) db.add(RunNode(run_id="run-1", node="sensor")) db.add(RunMetric(run_id="run-1", name="loss", step=-1)) db.add(RunArtifact(run_id="run-1", name="model.pt")) @@ -335,7 +335,12 @@ def test_delete_flow_is_refused_while_a_run_is_live( ) -> None: put_flow(client, superuser_token_headers, a_flow(), "demo") db.add( - Run(id="run-live", flow="demo", status="running", created_at=datetime.now(UTC)) + Run( + id="run-live", + flow_id="demo", + status="running", + created_at=datetime.now(UTC), + ) ) db.commit() @@ -506,30 +511,35 @@ def test_saving_a_header_moves_the_node_ports_in_the_draft( def test_a_shared_node_in_use_is_retired_only_with_detach( client: TestClient, superuser_token_headers: dict[str, str] ) -> None: - put_flow(client, superuser_token_headers, a_flow("shared"), "shared") + # Under an id that is not its key: the library names nodes by key. + put_flow(client, superuser_token_headers, a_flow("shared"), "x_shared") client.put( - f"{PREFIX}/shared/nodes/sensor/source", + f"{PREFIX}/x_shared/nodes/sensor/source", headers=superuser_token_headers, json={"code": WORKING_NODE}, ) client.post( - f"{PREFIX}/shared/nodes/sensor/share", + f"{PREFIX}/x_shared/nodes/sensor/share", headers=superuser_token_headers, json={"lib_name": "reading"}, ) + (shared,) = client.get(f"{PREFIX}/library", headers=superuser_token_headers).json() + assert shared["used_by"] == ["shared.sensor"] url = f"{PREFIX}/library/reading" - assert client.delete(url, headers=superuser_token_headers).status_code == 409 + refused = client.delete(url, headers=superuser_token_headers) + assert refused.status_code == 409 + assert "shared.sensor" in refused.json()["detail"] response = client.delete(f"{url}?detach=true", headers=superuser_token_headers) assert response.status_code == 200 assert client.get(f"{PREFIX}/library", headers=superuser_token_headers).json() == [] source = client.get( - f"{PREFIX}/shared/nodes/sensor/source", headers=superuser_token_headers + f"{PREFIX}/x_shared/nodes/sensor/source", headers=superuser_token_headers ) assert source.json()["code"] == WORKING_NODE - client.delete(f"{PREFIX}/shared", headers=superuser_token_headers) + client.delete(f"{PREFIX}/x_shared", headers=superuser_token_headers) def test_a_shared_node_only_the_published_flow_uses_is_kept( diff --git a/backend/tests/api/routes/test_history_ids.py b/backend/tests/api/routes/test_history_ids.py new file mode 100644 index 0000000..a698116 --- /dev/null +++ b/backend/tests/api/routes/test_history_ids.py @@ -0,0 +1,498 @@ +"""History by id, shown by key. + +A run, its nodes, its curve and its cache entry are filed under the flow's id +and named by the key the engine runs the flow under. Every test runs twice: on +a flow whose id is its key, and on one with an issued id, since a reader that +mixes the two up only shows on the second. +""" + +import csv +import io +import time +import uuid +from collections.abc import Callable +from datetime import UTC, datetime, timedelta +from typing import Any + +import pytest +from fastapi.testclient import TestClient +from sqlmodel import Session, col, delete, select + +from fluksio.core.config import settings +from fluksio.core.db import engine as db_engine +from fluksio.flow.events import EventBus +from fluksio.flow.metrics import MetricsCollector +from fluksio.models import ( + EngineEvent, + FlowRun, + MetricBucket, + Run, + RunMetric, + RunNode, + SQLModel, +) + +API = settings.API_V1_STR +FLOWS = f"{API}/flows" +RUNS = f"{API}/runs" + +#: A curve on the way, and a number at the end. +FIT = ( + "def process():\n" + " yield {'loss': 2.0}\n" + " yield {'loss': 1.0}\n" + " return {'final': 1.0}\n" +) + + +def a_batch(key: str) -> dict[str, Any]: + return { + "name": key, + "mode": "batch", + "outputs": ["final"], + "nodes": [ + { + "id": "fit", + "type": "python", + "provides": [ + {"name": "loss", "dtype": "float", "stream": True}, + {"name": "final", "dtype": "float"}, + ], + } + ], + } + + +Make = Callable[[str], str] + + +@pytest.fixture(params=["id_is_key", "issued"]) +def make( + request: pytest.FixtureRequest, + client: TestClient, + superuser_token_headers: dict[str, str], +) -> Make: + """Create and publish a batch flow, its id pinned to its key or issued.""" + headers = superuser_token_headers + made: list[str] = [] + + def create(key: str) -> str: + body = a_batch(key) + if request.param == "id_is_key": + body["id"] = key + response = client.post(f"{FLOWS}/", headers=headers, json=body) + assert response.status_code == 201, response.text + flow_id: str = response.json()["definition"]["id"] + assert (flow_id == key) is (request.param == "id_is_key") + made.append(flow_id) + client.put( + f"{FLOWS}/{flow_id}/nodes/fit/source", headers=headers, json={"code": FIT} + ) + version = client.get(f"{FLOWS}/{flow_id}", headers=headers).json()[ + "definition" + ]["version"] + published = client.post( + f"{FLOWS}/{flow_id}/publish", headers=headers, json={"version": version} + ) + assert published.status_code == 200, published.text + return flow_id + + yield create # type: ignore[misc] + for flow_id in made: + client.delete(f"{FLOWS}/{flow_id}", headers=headers) + client.delete(f"{RUNS}?flow={flow_id}&history=true", headers=headers) + + +def a_key() -> str: + return f"hist_{uuid.uuid4().hex[:8]}" + + +def run(client: TestClient, headers: dict[str, str], flow_id: str) -> dict[str, Any]: + """Submit a run over the API and wait for the engine to finish it.""" + submitted = client.post(f"{RUNS}/flows/{flow_id}", headers=headers, json={}) + assert submitted.status_code == 202, submitted.text + run_id = submitted.json()["id"] + deadline = time.monotonic() + 60 + while time.monotonic() < deadline: + detail: dict[str, Any] = client.get(f"{RUNS}/{run_id}", headers=headers).json() + if detail["status"] not in ("queued", "running"): + assert detail["status"] == "ok", detail["status_reason"] + return detail + time.sleep(0.1) + raise AssertionError(f"run {run_id} did not finish") + + +def test_a_batch_run_is_kept_by_id_and_named_by_key( + make: Make, client: TestClient, superuser_token_headers: dict[str, str] +) -> None: + """Submit, run, read back by id, export: the whole way for one flow.""" + headers = superuser_token_headers + key = a_key() + flow_id = make(key) + detail = run(client, headers, flow_id) + + assert (detail["flow"], detail["flow_name"]) == (flow_id, key) + assert detail["result"] == {"final": 1.0} + (node,) = detail["nodes"] + assert node["node"] == f"{key}.fit" + with Session(db_engine) as session: + assert session.get(Run, detail["id"]).flow_id == flow_id # type: ignore[union-attr] + stored = session.exec( + select(RunNode).where(col(RunNode.run_id) == detail["id"]) + ).one() + points = session.exec( + select(RunMetric).where(col(RunMetric.run_id) == detail["id"]) + ).all() + # The node by the flow's id, the message by its key. + assert stored.node == f"{flow_id}.fit" + assert {(p.node, p.name) for p in points} == {(f"{flow_id}.fit", f"{key}.loss")} + + outputs = client.get( + f"{RUNS}/{detail['id']}/nodes/{key}.fit/outputs", headers=headers + ) + assert outputs.json() == {f"{key}.final": 1.0} + curve = client.get(f"{RUNS}/{detail['id']}/metrics", headers=headers).json() + assert [(p["name"], p["value"]) for p in curve] == [ + (f"{key}.loss", 2.0), + (f"{key}.loss", 1.0), + ] + + # Filtered by id, strictly. The key is not a flow's id here unless it is. + (listed,) = client.get(RUNS, headers=headers, params={"flow": flow_id}).json() + assert (listed["flow"], listed["flow_name"]) == (flow_id, key) + by_key = client.get(RUNS, headers=headers, params={"flow": key}).json() + assert len(by_key) == (1 if flow_id == key else 0) + overview = client.get(f"{RUNS}/overview", headers=headers).json() + assert {"flow": flow_id, "flow_name": key} in [ + {"flow": row["flow"], "flow_name": row["flow_name"]} for row in overview + ] + + exported = client.get( + f"{RUNS}/export/runs", headers=headers, params={"flow": flow_id} + ).text + (row,) = csv.DictReader(io.StringIO(exported)) + assert (row["flow"], row["flow_name"], row["metric.final"]) == ( + flow_id, + key, + "1.0", + ) + series = client.get( + f"{RUNS}/export/metrics", + headers=headers, + params={"flow": flow_id, "format": "jsonl"}, + ).text + assert f'"name": "{key}.loss"' in series + + # The cascade the collector made of it, by id as well. + client.post(f"{API}/observability/flush", headers=headers) + cascades = client.get( + f"{API}/observability/runs", headers=headers, params={"flow": flow_id} + ).json()["data"] + assert [(c["id"], c["flow"], c["flow_name"]) for c in cascades] == [ + (detail["id"], flow_id, key) + ] + + # Gone, the history stays under the id and keeps the key it last had. + client.delete(f"{FLOWS}/{flow_id}", headers=headers) + (kept,) = client.get(RUNS, headers=headers, params={"flow": flow_id}).json() + assert (kept["flow"], kept["flow_name"]) == (flow_id, key) + again = client.get(f"{RUNS}/{detail['id']}", headers=headers).json() + assert again["nodes"][0]["node"] == f"{key}.fit" + kept_outputs = client.get( + f"{RUNS}/{detail['id']}/nodes/{key}.fit/outputs", headers=headers + ) + assert kept_outputs.json() == {f"{key}.final": 1.0} + + +def test_a_cache_entry_is_restored_into_another_flow_by_its_key( + make: Make, client: TestClient, superuser_token_headers: dict[str, str] +) -> None: + """The same node in two flows: the second run restores the first's. + + Outputs are named by the source flow's key and restored under the asking + flow's. The curve is read from the source run, its rows found by the + source flow's id, and renamed between the two keys. Deleting the source + flow changes neither. + """ + headers = superuser_token_headers + source_key, reuse_key = a_key(), a_key() + source_id = make(source_key) + first = run(client, headers, source_id) + reuse_id = make(reuse_key) + reused = run(client, headers, reuse_id) + + (node,) = reused["nodes"] + assert (node["node"], node["status"], node["cached_from"]) == ( + f"{reuse_key}.fit", + "cached", + first["id"], + ) + assert reused["result"] == {"final": 1.0} + outputs = client.get( + f"{RUNS}/{reused['id']}/nodes/{reuse_key}.fit/outputs", headers=headers + ).json() + assert outputs == {f"{reuse_key}.final": 1.0} + + def curve() -> list[tuple[str, float]]: + points = client.get(f"{RUNS}/{reused['id']}/metrics", headers=headers).json() + return [(p["name"], p["value"]) for p in points] + + assert curve() == [(f"{reuse_key}.loss", 2.0), (f"{reuse_key}.loss", 1.0)] + compared = client.get( + f"{RUNS}/series/compare", + headers=headers, + params={"ids": f"{first['id']},{reused['id']}", "metric": f"{reuse_key}.loss"}, + ).json() + assert [len(line["points"]) for line in compared["lines"]] == [0, 2] + + # The source flow goes, its run stays, and so does what it holds. + client.delete(f"{FLOWS}/{source_id}", headers=headers) + assert curve() == [(f"{reuse_key}.loss", 2.0), (f"{reuse_key}.loss", 1.0)] + with Session(db_engine) as session: + cache_key = session.exec( + select(RunNode.cache_key).where(col(RunNode.run_id) == first["id"]) + ).one() + hit = client.app.state.run_service._cache.lookup(cache_key) # type: ignore[attr-defined] + # The newest entry is the restored one, named by the flow that restored it. + assert hit is not None and hit.flow == reuse_key + with Session(db_engine) as session: + for row in session.exec( + select(RunNode).where(col(RunNode.run_id) == reused["id"]) + ): + row.cache_key = "" + session.add(row) + session.commit() + hit = client.app.state.run_service._cache.lookup(cache_key) # type: ignore[attr-defined] + # And past it, the source run's, under the key its flow had when it went. + assert hit is not None and hit.flow == source_key + assert hit.outputs == {f"{source_key}.final": 1.0} + + +def test_engine_history_is_read_by_id_and_names_nodes_by_key( + make: Make, client: TestClient, superuser_token_headers: dict[str, str] +) -> None: + """Rollups and failures, filtered by the flow's id, shown by its key.""" + headers = superuser_token_headers + key = a_key() + flow_id = make(key) + minute = datetime.now(UTC).replace(second=0, microsecond=0) - timedelta(minutes=2) + with Session(db_engine) as session: + session.add( + MetricBucket( + flow_id=flow_id, node=f"{flow_id}.fit", bucket=minute, executions=3 + ) + ) + session.add( + EngineEvent( + ts=minute, type="node_error", flow_id=flow_id, node=f"{flow_id}.fit" + ) + ) + session.commit() + try: + (event,) = client.get( + f"{API}/observability/events", headers=headers, params={"flow": flow_id} + ).json() + assert (event["flow"], event["flow_name"], event["node"]) == ( + flow_id, + key, + f"{key}.fit", + ) + points = client.get( + f"{API}/observability/timeseries", + headers=headers, + params={"flow": flow_id, "node": f"{key}.fit"}, + ).json() + assert sum(point["executions"] for point in points) == 3 + rollups = client.get(f"{API}/observability/flows", headers=headers).json() + (rollup,) = [row for row in rollups if row["flow"] == flow_id] + assert (rollup["flow_name"], rollup["executions"]) == (key, 3) + finally: + with Session(db_engine) as session: + session.execute( + delete(MetricBucket).where(col(MetricBucket.flow_id) == flow_id) + ) + session.execute( + delete(EngineEvent).where(col(EngineEvent.flow_id) == flow_id) + ) + session.commit() + + +def test_events_carry_the_flow_id_and_rows_are_filed_under_it() -> None: + """The id is attached on publish, before anything buffers the event.""" + bus = EventBus() + bus.flow_id_of = {"heat": "f_heat0001"}.get # type: ignore[assignment] + collector = MetricsCollector(bus) + events = [ + {"type": "node_executed", "flow": "heat", "node": "heat.pump", "ts": 60.0}, + {"type": "node_error", "flow": "heat", "node": "heat.pump", "ts": 60.0}, + {"type": "engine_degraded", "flow": "", "reason": "lag"}, + ] + for event in events: + bus.publish(event) + collector.handle(event) + + # What every subscriber is handed, the socket and the collector alike. + assert [e.get("flow_id") for e in events] == ["f_heat0001", "f_heat0001", ""] + ((flow_id, node, _),) = collector._buckets + assert (flow_id, node) == ("f_heat0001", "f_heat0001.pump") + error, degraded = collector._pending + assert (error.flow_id, error.node) == ("f_heat0001", "f_heat0001.pump") + # The engine as a whole stays nobody's. + assert (degraded.flow_id, degraded.node) == ("", "") + + +# ----------------------------------------------------------------------------- +# Registry guard +# +# Every column of the history tables is classified by what it names. A new +# column, or a new table, fails here until somebody says whether it holds a +# flow id, a node by flow id, a message by key or nothing of either: the one +# place where history could start naming a flow by key again unnoticed. +# ----------------------------------------------------------------------------- + +FLOW_ID = "flow id" +NODE = "." +MESSAGE = "message by key" +PAYLOAD = "payload or local names, never prefixed with an id" +OTHER = "names no flow" + +HISTORY: dict[str, dict[str, str]] = { + "run": { + "flow": FLOW_ID, + # A training run's target, in a flow other than the run's. + "for_node": NODE, + "params": PAYLOAD, + "passed": PAYLOAD, + "declared": PAYLOAD, + "result": PAYLOAD, + "status_reason": PAYLOAD, + **dict.fromkeys( + ( + "id", + "flow_version", + "commit", + "origin_commit", + "code_digest", + "params_digest", + "seed", + "group_id", + "parent_id", + "idempotency_key", + "external_id", + "cause", + "no_cache", + "draft", + "status", + "labels", + "needs", + "created_at", + "started_at", + "finished_at", + "duration_ms", + "engine", + "lease_at", + "actor", + ), + OTHER, + ), + }, + "run_node": { + "node": NODE, + # The keys of the JSON object are messages; the values are payload. + "outputs": MESSAGE, + "logs": PAYLOAD, + "error": PAYLOAD, + **dict.fromkeys( + ( + "run_id", + "status", + "attempt", + "started_at", + "duration_ms", + "worker", + "cache_key", + "cached_from", + ), + OTHER, + ), + }, + "run_metric": { + "node": NODE, + "name": MESSAGE, + **dict.fromkeys(("run_id", "step", "ts", "value"), OTHER), + }, + "run_artifact": { + "node": NODE, + "name": MESSAGE, + "filename": PAYLOAD, + **dict.fromkeys( + ("run_id", "digest", "size", "media_type", "created_at"), OTHER + ), + }, + "flow_run": { + "flow": FLOW_ID, + **dict.fromkeys( + ( + "id", + "source", + "parent", + "started_at", + "finished_at", + "status", + "nodes", + "errors", + "duration_ms", + "deliveries", + ), + OTHER, + ), + }, + "metric_minute": { + "flow": FLOW_ID, + "node": NODE, + **dict.fromkeys( + ( + "bucket", + "executions", + "errors", + "messages", + "duration_sum_ms", + "duration_max_ms", + "lag_sum_ms", + "lag_max_ms", + "items", + ), + OTHER, + ), + }, + "engine_event": { + "flow": FLOW_ID, + "node": NODE, + "detail": PAYLOAD, + **dict.fromkeys(("id", "ts", "type", "actor", "run"), OTHER), + }, + # `:`, moved by a rename like a message. + "alert_sent": {"key": MESSAGE, "id": OTHER, "ts": OTHER}, + "flow_tombstone": {"id": FLOW_ID, "name": PAYLOAD, "deleted_at": OTHER}, +} + +#: Tables that hold no flow history at all. +UNRELATED = { + "user", + "oauth_client", + "oauth_authorization_code", + "oauth_refresh_token", + "flavor", + "connector_state", + "worker_token", +} + + +def test_every_history_column_says_what_it_names() -> None: + tables = SQLModel.metadata.tables + assert set(tables) == set(HISTORY) | UNRELATED + for name, columns in HISTORY.items(): + assert set(tables[name].columns.keys()) == set(columns), name + # The identity columns keep their name in SQL and read as ids in Python. + for model in (Run, FlowRun, MetricBucket, EngineEvent): + assert hasattr(model, "flow_id") and not hasattr(model, "flow") diff --git a/backend/tests/api/routes/test_observability.py b/backend/tests/api/routes/test_observability.py index 5a56b42..3d23088 100644 --- a/backend/tests/api/routes/test_observability.py +++ b/backend/tests/api/routes/test_observability.py @@ -16,7 +16,7 @@ def _seed(db: Session) -> None: now = datetime.now(UTC).replace(second=0, microsecond=0) db.add( MetricBucket( - flow=FLOW, + flow_id=FLOW, node=f"{FLOW}.calc", bucket=now - timedelta(minutes=1), executions=4, @@ -33,20 +33,24 @@ def _seed(db: Session) -> None: EngineEvent( ts=now, type="node_error", - flow=FLOW, + flow_id=FLOW, node=f"{FLOW}.calc", detail="ValueError: bad input\nTraceback", ) ) db.add( EngineEvent( - ts=now, type="audit", flow=FLOW, detail="published", actor="a@example.com" + ts=now, + type="audit", + flow_id=FLOW, + detail="published", + actor="a@example.com", ) ) db.add( FlowRun( id="9-0", - flow=FLOW, + flow_id=FLOW, source="external", started_at=now, finished_at=now, @@ -208,7 +212,7 @@ def test_the_timeseries_folds_into_the_requested_bucket( for offset, executions in ((0, 1), (2, 2), (5, 4), (15, 8)): db.add( MetricBucket( - flow=flow, + flow_id=flow, node=f"{flow}.calc", bucket=first + timedelta(minutes=offset), executions=executions, @@ -242,7 +246,7 @@ def test_the_bucket_still_filling_is_held_back( for offset, executions in ((-60, 3), (0, 1)): db.add( MetricBucket( - flow=flow, + flow_id=flow, node=f"{flow}.calc", bucket=datetime.fromtimestamp(minute + offset, UTC), executions=executions, @@ -289,7 +293,7 @@ def test_the_runs_page_carries_its_total( db.add( FlowRun( id=f"paged-{index}", - flow=FLOW, + flow_id=FLOW, started_at=minute + timedelta(seconds=index), status="ok", ) @@ -315,11 +319,11 @@ def test_runs_narrow_to_one_minute( ) -> None: """A minute picked off a chart reaches past what the recent list holds.""" minute = datetime.now(UTC).replace(second=0, microsecond=0) - timedelta(hours=3) - db.add(FlowRun(id="minute-in", flow=FLOW, started_at=minute, status="ok")) + db.add(FlowRun(id="minute-in", flow_id=FLOW, started_at=minute, status="ok")) db.add( FlowRun( id="minute-after", - flow=FLOW, + flow_id=FLOW, started_at=minute + timedelta(minutes=1), status="ok", ) @@ -345,7 +349,7 @@ def test_a_running_cascade_reports_how_long_it_has_been_going( ) -> None: """The collector writes a duration at the end; until then, time since.""" started = datetime.now(UTC) - timedelta(seconds=30) - db.add(FlowRun(id="in-flight", flow=FLOW, started_at=started, status="running")) + db.add(FlowRun(id="in-flight", flow_id=FLOW, started_at=started, status="running")) db.commit() runs = client.get( @@ -363,12 +367,12 @@ def test_events_narrow_to_one_minute( ) -> None: """The failures list reaches as far back as the charts beside it.""" minute = datetime.now(UTC).replace(second=0, microsecond=0) - timedelta(hours=4) - db.add(EngineEvent(ts=minute, type="node_error", flow=FLOW, detail="minute-in")) + db.add(EngineEvent(ts=minute, type="node_error", flow_id=FLOW, detail="minute-in")) db.add( EngineEvent( ts=minute + timedelta(minutes=1), type="node_error", - flow=FLOW, + flow_id=FLOW, detail="minute-after", ) ) @@ -394,16 +398,18 @@ def test_events_narrow_to_one_run( """Reading one run's failures without filtering the engine-wide list.""" ts = datetime.now(UTC) - timedelta(hours=6) db.add( - EngineEvent(ts=ts, type="node_error", flow=FLOW, detail="mine", run="run-mine") + EngineEvent( + ts=ts, type="node_error", flow_id=FLOW, detail="mine", run="run-mine" + ) ) db.add( EngineEvent( - ts=ts, type="node_error", flow=FLOW, detail="theirs", run="run-theirs" + ts=ts, type="node_error", flow_id=FLOW, detail="theirs", run="run-theirs" ) ) # What a live cascade unrelated to any run leaves, and what every row # written before the column existed looks like. - db.add(EngineEvent(ts=ts, type="node_error", flow=FLOW, detail="neither")) + db.add(EngineEvent(ts=ts, type="node_error", flow_id=FLOW, detail="neither")) db.commit() events = client.get( @@ -423,7 +429,7 @@ def test_a_timeseries_window_can_be_named( for minute in range(4): db.add( MetricBucket( - flow="windowed", + flow_id="windowed", node="windowed.n", bucket=base + timedelta(minutes=minute), executions=1, @@ -553,7 +559,7 @@ def test_a_superuser_can_flush_the_collector_now( written = db.exec( select(EngineEvent).where( - EngineEvent.flow == FLOW, EngineEvent.detail == "flushed" + EngineEvent.flow_id == FLOW, EngineEvent.detail == "flushed" ) ).all() assert len(written) == 1 @@ -564,12 +570,14 @@ def test_execution_time_is_summed_per_node_type( ) -> None: from fluksio.flow.schemas import FlowDef, NodeDef - flow = "node_type_test" + # Rows name a node by its flow's id, which here is not its key. + flow, key = "x_node_type_test", "node_type_test" controller = client.app.state.flow_controller before = controller.store.read_all controller.store.read_all = lambda: [ FlowDef( - name=flow, + id=flow, + name=key, nodes=[ NodeDef(id="a", type="python"), NodeDef(id="b", type="python"), @@ -587,7 +595,7 @@ def test_execution_time_is_summed_per_node_type( ): db.add( MetricBucket( - flow=flow, + flow_id=flow, node=f"{flow}.{node}", bucket=now - timedelta(minutes=2), executions=executions, diff --git a/backend/tests/api/routes/test_runs.py b/backend/tests/api/routes/test_runs.py index 0f1d23c..55b7186 100644 --- a/backend/tests/api/routes/test_runs.py +++ b/backend/tests/api/routes/test_runs.py @@ -45,7 +45,12 @@ def test_a_run_cache_finds_what_an_earlier_run_recorded(tmp_path): with Session(db_engine) as session: for run_id in ("cache-1", "cache-2", "cache-3"): session.add( - Run(id=run_id, flow="study", status="ok", created_at=datetime.now(UTC)) + Run( + id=run_id, + flow_id="study", + status="ok", + created_at=datetime.now(UTC), + ) ) session.add( RunNode( @@ -115,7 +120,7 @@ def test_a_row_keyed_before_the_current_shape_is_unkeyed(): ("shape-old", CACHE_SHAPE_SINCE - timedelta(days=1)), ("shape-new", CACHE_SHAPE_SINCE + timedelta(seconds=1)), ): - session.add(Run(id=run_id, flow="study", status="ok", created_at=made)) + session.add(Run(id=run_id, flow_id="study", status="ok", created_at=made)) session.add( RunNode( run_id=run_id, @@ -174,7 +179,7 @@ def made_artifact(): session.add( Run( id=run_id, - flow="prepare", + flow_id="prepare", status="ok", result={"dataset": reference}, created_at=datetime.now(UTC), @@ -288,7 +293,7 @@ def made_config(): session.add( Run( id=run_id, - flow="generate", + flow_id="generate", status="ok", result={"meta": meta, "label": "run-7"}, created_at=datetime.now(UTC), @@ -335,7 +340,7 @@ def test_the_overview_counts_a_flow_the_list_page_would_not_reach( session.add( Run( id=f"ov-{index}", - flow="overviewed", + flow_id="overviewed", status="ok" if index else "running", created_at=datetime(2026, 1, 1 + index, tzinfo=UTC), ) @@ -358,7 +363,7 @@ def test_a_running_run_reports_how_long_it_has_been_going( session.add( Run( id="in-flight", - flow="timed", + flow_id="timed", status="running", created_at=datetime.now(UTC), started_at=datetime.now(UTC) - timedelta(seconds=30), @@ -390,7 +395,7 @@ def test_a_run_records_which_caller_asked_for_it( seen.update(kwargs) return Run( id="cause-1", - flow=name, + flow_id=name, cause=str(kwargs["cause"]), created_at=datetime.now(UTC), ) @@ -431,7 +436,7 @@ def test_a_repeated_submit_returns_the_run_it_already_made(): session.add( Run( id="dedup-1", - flow="study", + flow_id="study", status="running", idempotency_key="key-abc", created_at=datetime.now(UTC), @@ -449,11 +454,15 @@ class _OneFlow: def __init__(self, flow): self.store = self - self._flow = flow + # As the store reads one from before ids: its key is its id. + self._flow = flow.model_copy(update={"id": flow.id or flow.name}) def read_flow(self, name, draft=False): return self._flow + def key_index(self): + return {self._flow.name: self._flow.id} + def head(self): return "" @@ -738,7 +747,7 @@ def test_a_retry_is_a_new_run_that_names_the_one_it_repeats(): session.add( Run( id="abandoned-1", - flow="study", + flow_id="study", status="abandoned", params={"lr": 0.3}, group_id="sweep-9", @@ -749,7 +758,7 @@ def test_a_retry_is_a_new_run_that_names_the_one_it_repeats(): session.add( Run( id="going-1", - flow="study", + flow_id="study", status="running", created_at=datetime.now(UTC), ) @@ -796,7 +805,7 @@ def test_runs_left_queued_are_woken_by_the_next_engine(): session.add( Run( id="orphan-1", - flow="study", + flow_id="study", status="queued", created_at=datetime.now(UTC), ) @@ -804,7 +813,7 @@ def test_runs_left_queued_are_woken_by_the_next_engine(): session.add( Run( id="finished-1", - flow="study", + flow_id="study", status="ok", created_at=datetime.now(UTC), ) @@ -903,7 +912,7 @@ def test_a_key_nobody_used_submits_normally( class Recorder: def submit(self, name, **kwargs): seen.update(kwargs) - return Run(id="keyed-1", flow=name, created_at=datetime.now(UTC)) + return Run(id="keyed-1", flow_id=name, created_at=datetime.now(UTC)) monkeypatch.setattr(client.app.state, "run_service", Recorder()) answer = client.post( @@ -940,7 +949,7 @@ def deletable_run(): run_id = "del-1" with Session(db_engine) as session: session.add( - Run(id=run_id, flow="deleted", status="ok", created_at=datetime.now(UTC)) + Run(id=run_id, flow_id="deleted", status="ok", created_at=datetime.now(UTC)) ) session.add(RunNode(run_id=run_id, node="deleted.a", status="ok")) session.add(RunMetric(run_id=run_id, name="deleted.loss", step=0, value=1.0)) @@ -985,7 +994,11 @@ def test_a_node_finishing_after_its_run_was_deleted_is_dropped_quietly(caplog): """The constraint refuses its row, and that is a no-op rather than an error.""" service = RunService(controller=_OneFlow(FlowDef(name="gone")), queue=_Collect()) with caplog.at_level(logging.INFO, logger="fluksio.flow.runs"): - service._record_node("del-gone", NodeOutcome(node="gone.a", ok=True)) + service._record_node( + Run(id="del-gone", flow_id="gone"), + "gone", + NodeOutcome(node="gone.a", ok=True), + ) with Session(db_engine) as session: assert session.get(RunNode, {"run_id": "del-gone", "node": "gone.a"}) is None @@ -1008,7 +1021,7 @@ def test_a_running_run_is_refused_rather_than_raced(client, superuser_token_head session.add( Run( id=run_id, - flow="deleted", + flow_id="deleted", status="running", created_at=datetime.now(UTC), ) @@ -1041,11 +1054,13 @@ def test_deleting_a_flows_runs_clears_that_flow_only(client, superuser_token_hea made = datetime.now(UTC) url = f"{settings.API_V1_STR}/runs" with Session(db_engine) as session: - session.add(Run(id="bulk-1", flow="swept", status="ok", created_at=made)) + session.add(Run(id="bulk-1", flow_id="swept", status="ok", created_at=made)) session.add(RunNode(run_id="bulk-1", node="swept.a", status="ok")) session.add(RunMetric(run_id="bulk-1", name="swept.loss", step=0, value=1.0)) - session.add(Run(id="bulk-2", flow="swept", status="running", created_at=made)) - session.add(Run(id="keep-1", flow="other", status="ok", created_at=made)) + session.add( + Run(id="bulk-2", flow_id="swept", status="running", created_at=made) + ) + session.add(Run(id="keep-1", flow_id="other", status="ok", created_at=made)) session.commit() busy = client.delete(url, params={"flow": "swept"}, headers=superuser_token_headers) @@ -1063,7 +1078,7 @@ def test_deleting_a_flows_runs_clears_that_flow_only(client, superuser_token_hea assert "2" in answer.json()["message"] with Session(db_engine) as session: - assert session.exec(select(Run).where(col(Run.flow) == "swept")).all() == [] + assert session.exec(select(Run).where(col(Run.flow_id) == "swept")).all() == [] for table in (RunNode, RunMetric): assert ( session.exec(select(table).where(col(table.run_id) == "bulk-1")).all() @@ -1079,10 +1094,10 @@ def test_history_also_clears_a_flows_cascades(client, superuser_token_headers): made = datetime.now(UTC) url = f"{settings.API_V1_STR}/runs" with Session(db_engine) as session: - session.add(Run(id="hist-1", flow="gone", status="ok", created_at=made)) - session.add(FlowRun(id="hist-1", flow="gone", source="run", started_at=made)) - session.add(FlowRun(id="hist-2", flow="gone", started_at=made)) - session.add(FlowRun(id="hist-3", flow="other", started_at=made)) + session.add(Run(id="hist-1", flow_id="gone", status="ok", created_at=made)) + session.add(FlowRun(id="hist-1", flow_id="gone", source="run", started_at=made)) + session.add(FlowRun(id="hist-2", flow_id="gone", started_at=made)) + session.add(FlowRun(id="hist-3", flow_id="other", started_at=made)) session.commit() plain = client.delete(url, params={"flow": "gone"}, headers=superuser_token_headers) @@ -1095,7 +1110,7 @@ def test_history_also_clears_a_flows_cascades(client, superuser_token_headers): ) assert answer.status_code == 200 with Session(db_engine) as session: - left = session.exec(select(FlowRun).where(col(FlowRun.flow) == "gone")).all() + left = session.exec(select(FlowRun).where(col(FlowRun.flow_id) == "gone")).all() assert left == [] session.delete(session.get(FlowRun, "hist-3")) session.commit() @@ -1114,8 +1129,8 @@ def reused_run(): """A run of `quick` whose node was restored from a run of `train`.""" with Session(db_engine) as session: made = datetime.now(UTC) - session.add(Run(id="src-1", flow="train", status="ok", created_at=made)) - session.add(Run(id="reuse-1", flow="quick", status="ok", created_at=made)) + session.add(Run(id="src-1", flow_id="train", status="ok", created_at=made)) + session.add(Run(id="reuse-1", flow_id="quick", status="ok", created_at=made)) session.add(RunNode(run_id="src-1", node="train.fit", status="ok")) session.add( RunNode( @@ -1197,7 +1212,7 @@ def test_a_node_answers_with_what_it_returned(client, superuser_token_headers): run_id = new_run_id() with Session(db_engine) as session: session.add( - Run(id=run_id, flow="study", status="ok", created_at=datetime.now(UTC)) + Run(id=run_id, flow_id="study", status="ok", created_at=datetime.now(UTC)) ) session.add(RunNode(run_id=run_id, node="fit", outputs='{"study.loss":0.5}')) session.add(RunNode(run_id=run_id, node="plot", status="error")) @@ -1233,7 +1248,7 @@ def plotted(): run_id = new_run_id() with Session(db_engine) as session: session.add( - Run(id=run_id, flow="study", status="ok", created_at=datetime.now(UTC)) + Run(id=run_id, flow_id="study", status="ok", created_at=datetime.now(UTC)) ) for step, (loss, epoch) in enumerate([(1.0, 10.0), (0.5, 20.0), (0.25, 30.0)]): session.add(_metric(run_id, "study.loss", step, loss, 100.0 + step * 5)) @@ -1263,7 +1278,7 @@ def test_the_metric_names_of_a_selection_are_exact( session.add( Run( id=older, - flow="study", + flow_id="study", status="ok", created_at=datetime.now(UTC) - timedelta(hours=1), ) @@ -1374,7 +1389,7 @@ def test_a_run_records_the_code_it_started_with_not_the_code_it_was_queued_with( session.add( Run( id="stamp-1", - flow="study", + flow_id="study", status="queued", code_digest="at-submit", created_at=datetime.now(UTC), @@ -1395,7 +1410,7 @@ def test_a_tree_that_did_not_move_is_not_written_again(): session.add( Run( id="stamp-2", - flow="study", + flow_id="study", status="queued", code_digest="same", created_at=datetime.now(UTC), @@ -1429,7 +1444,7 @@ def exported(): session.add( Run( id=run_id, - flow="export-study", + flow_id="export-study", status="ok", params={ "lr": lr, @@ -1531,7 +1546,7 @@ def test_an_exported_run_row_carries_every_recorded_input( # 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.startswith("id,flow,flow_name,status,") assert header.endswith( "param.config.depth,param.config.model,param.epochs,param.lr," "metric.acc,metric.final_metrics.train_loss," @@ -1557,7 +1572,7 @@ def paged(): session.add( Run( id=f"page-{index}", - flow="paged-study", + flow_id="paged-study", status="ok", created_at=made + timedelta(minutes=index), ) diff --git a/backend/tests/api/routes/test_search.py b/backend/tests/api/routes/test_search.py index 459bc09..a07583a 100644 --- a/backend/tests/api/routes/test_search.py +++ b/backend/tests/api/routes/test_search.py @@ -31,7 +31,8 @@ def test_index_reaches_inside_flows_and_dashboards( "title": "Searchable", "nodes": [{"id": "sensor", "type": "python", "title": "Hall sensor"}], }, - "searchable", + # Ids that are not the keys, which is what a result opens by. + "x_searchable", ) put_dashboard( client, @@ -50,7 +51,7 @@ def test_index_reaches_inside_flows_and_dashboards( ], "version": 0, }, - "hall", + "x_hall", ) entries = client.get(f"{PREFIX}/", headers=superuser_token_headers).json() @@ -66,6 +67,17 @@ def test_index_reaches_inside_flows_and_dashboards( assert found[("dashboard", "", "hall")]["title"] == "Hall" assert found[("widget", "hall", "temperature")]["title"] == "Temperature" assert found[("widget", "hall", "temperature")]["kind"] == "stat" + assert [ + found[entry]["id"] + for entry in ( + ("flow", "", "searchable"), + ("node", "searchable", "sensor"), + ("dashboard", "", "hall"), + ("widget", "hall", "temperature"), + ) + ] == ["x_searchable", "x_searchable", "x_hall", "x_hall"] + client.delete(f"{FLOWS}/x_searchable", headers=superuser_token_headers) + client.delete(f"{DASHBOARDS}/x_hall", headers=superuser_token_headers) def test_index_lists_a_flows_inputs_and_outputs( diff --git a/backend/tests/api/routes/test_wren_train.py b/backend/tests/api/routes/test_wren_train.py index cb53972..e049f97 100644 --- a/backend/tests/api/routes/test_wren_train.py +++ b/backend/tests/api/routes/test_wren_train.py @@ -86,15 +86,28 @@ def test_train_submits_the_node_as_wren_s_installation( class Recorder: def submit(self, name, **kwargs): seen.update(kwargs, flow=name) - return Run(id="train-1", flow=name, created_at=datetime.now(UTC)) + return Run(id="train-1", flow_id=name, created_at=datetime.now(UTC)) + # Registered by sync, so under an id of its own: found by its key. + registered = put_flow( + client, + superuser_token_headers, + {"name": "wren_train", "mode": "batch"}, + "x_wren_train", + ) + assert registered.is_success, registered.text monkeypatch.setattr(client.app.state, "run_service", Recorder()) - answer = client.post(f"{PREFIX}/control/train", headers=superuser_token_headers) + try: + answer = client.post(f"{PREFIX}/control/train", headers=superuser_token_headers) + finally: + client.delete( + f"{settings.API_V1_STR}/flows/x_wren_train", headers=superuser_token_headers + ) assert answer.status_code == 202, answer.text assert answer.json()["id"] == "train-1" assert (seen["flow"], seen["no_cache"], seen["cause"], seen["for_node"]) == ( - "wren_train", + "x_wren_train", True, "train", f"{FLOW}.control", @@ -161,7 +174,7 @@ def test_a_node_reads_its_own_latest_training( session.add( Run( id=run_id, - flow="wren_train", + flow_id="wren_train", status="ok", for_node=f"{FLOW}.{node}", result={"exact_match": 0.98, "coverage": 0.9}, diff --git a/backend/tests/test_artifacts.py b/backend/tests/test_artifacts.py index cbb5068..afa7736 100644 --- a/backend/tests/test_artifacts.py +++ b/backend/tests/test_artifacts.py @@ -28,7 +28,7 @@ def _run_row(status: str, digest: str = "") -> str: run_id = new_run_id() with Session(db_engine) as session: session.add( - Run(id=run_id, flow="f", status=status, created_at=datetime.now(UTC)) + Run(id=run_id, flow_id="f", status=status, created_at=datetime.now(UTC)) ) if digest: session.add( diff --git a/backend/tests/test_db.py b/backend/tests/test_db.py index fc3aad6..3440564 100644 --- a/backend/tests/test_db.py +++ b/backend/tests/test_db.py @@ -20,12 +20,14 @@ def test_a_stored_instant_comes_back_aware_and_in_utc() -> None: stamp = datetime(2026, 8, 21, 15, 30, tzinfo=timezone(timedelta(hours=2))) with Session(engine) as session: session.add( - EngineEvent(ts=stamp, type="node_error", flow="tz", node="n", detail="") + EngineEvent(ts=stamp, type="node_error", flow_id="tz", node="n", detail="") ) session.commit() with Session(engine) as session: - stored = session.exec(select(EngineEvent).where(EngineEvent.flow == "tz")).one() + stored = session.exec( + select(EngineEvent).where(EngineEvent.flow_id == "tz") + ).one() assert stored.ts.tzinfo is not None assert stored.ts.utcoffset() == timedelta(0) assert stored.ts == stamp diff --git a/backend/tests/test_metrics.py b/backend/tests/test_metrics.py index d1e5fe9..e2d90c6 100644 --- a/backend/tests/test_metrics.py +++ b/backend/tests/test_metrics.py @@ -93,7 +93,9 @@ def test_events_become_rollups_failures_runs_and_audit(db: Session) -> None: asyncio.run(collector.flush()) bucket = db.exec( - select(MetricBucket).where(MetricBucket.flow == FLOW, MetricBucket.node == NODE) + select(MetricBucket).where( + MetricBucket.flow_id == FLOW, MetricBucket.node == NODE + ) ).one() assert (bucket.executions, bucket.errors, bucket.messages) == (1, 1, 2) assert bucket.duration_max_ms == 5.0 @@ -101,19 +103,21 @@ def test_events_become_rollups_failures_runs_and_audit(db: Session) -> None: failure = db.exec( select(EngineEvent).where( - EngineEvent.flow == FLOW, EngineEvent.type == "node_error" + EngineEvent.flow_id == FLOW, EngineEvent.type == "node_error" ) ).one() assert "ValueError: bad input" in failure.detail assert "line 3, in run" in failure.detail audit = db.exec( - select(EngineEvent).where(EngineEvent.flow == FLOW, EngineEvent.type == "audit") + select(EngineEvent).where( + EngineEvent.flow_id == FLOW, EngineEvent.type == "audit" + ) ).one() assert (audit.actor, audit.detail) == ("someone@example.com", "published") run = db.exec(select(FlowRun).where(FlowRun.id == "1-0")).one() - assert (run.status, run.flow, run.source) == ("error", FLOW, "external") + assert (run.status, run.flow_id, run.source) == ("error", FLOW, "external") assert (run.nodes, run.errors) == (1, 1) assert run.duration_ms > 0 @@ -122,7 +126,9 @@ def test_events_become_rollups_failures_runs_and_audit(db: Session) -> None: asyncio.run(collector.flush()) db.expire_all() bucket = db.exec( - select(MetricBucket).where(MetricBucket.flow == FLOW, MetricBucket.node == NODE) + select(MetricBucket).where( + MetricBucket.flow_id == FLOW, MetricBucket.node == NODE + ) ).one() assert (bucket.executions, bucket.errors) == (2, 2) @@ -196,7 +202,7 @@ def test_batched_rollups_add_and_redelivery_clears_a_finished_run(db: Session) - bucket["duration_max_ms"] = 4.0 collector._write(collector._buckets, [], list(collector._runs.values()), False) - buckets = db.exec(select(MetricBucket).where(MetricBucket.flow == "bulk")).all() + buckets = db.exec(select(MetricBucket).where(MetricBucket.flow_id == "bulk")).all() assert len(buckets) == 2 assert all( (b.executions, b.messages, b.duration_sum_ms, b.duration_max_ms) @@ -400,4 +406,4 @@ def test_a_rate_limit_flush_is_a_run_that_names_its_parent(db: Session) -> None: assert (flush.status, flush.parent) == ("ok", held_in) parent = db.get(FlowRun, held_in) assert parent is not None - assert (parent.flow, parent.source) == ("metrics-flush", "external") + assert (parent.flow_id, parent.source) == ("metrics-flush", "external") diff --git a/backend/tests/test_tombstones.py b/backend/tests/test_tombstones.py index 8e6c5c0..f9fd284 100644 --- a/backend/tests/test_tombstones.py +++ b/backend/tests/test_tombstones.py @@ -22,14 +22,14 @@ def test_a_tombstone_stays_while_history_names_its_id(db: Session) -> None: db.add( Run( id="tomb-run", - flow="wren_train", + flow_id="wren_train", for_node="f_trainee.lights", created_at=now, ) ) - db.add(Run(id="tomb-own", flow="f_gonegone", created_at=now)) - db.add(MetricBucket(flow="other", node="f_measured.sensor", bucket=now)) - db.add(EngineEvent(ts=now, type="audit", flow="f_forgotten_too")) + db.add(Run(id="tomb-own", flow_id="f_gonegone", created_at=now)) + db.add(MetricBucket(flow_id="other", node="f_measured.sensor", bucket=now)) + db.add(EngineEvent(ts=now, type="audit", flow_id="f_forgotten_too")) db.commit() assert graves.recorded("f_trainee") == "trainee" diff --git a/docs/code/api.md b/docs/code/api.md index a28b08f..24fe495 100644 --- a/docs/code/api.md +++ b/docs/code/api.md @@ -86,6 +86,12 @@ Deleting a flow keeps its run history, which is shown under the key the flow last had. Delete runs separately. The flow's values and queued work go with it, so its key is free for a new flow, which starts with an empty history. +Answers that lead to a flow or dashboard carry its id next to the key. An +endpoint in `GET /flows/{id}` has `owner_id`, a neuron in `GET /flows/graph` +lists `flow_ids` in the order of its `flows`, a device binding has `flow_id` +and a search entry has `id`. Open things by these ids, not by the part of a +`.` before the dot. + ### Node source ```sh @@ -106,7 +112,8 @@ Ordinary source edits stay in the draft until publishing. Shared library source changes affect its users immediately, and the audit trail records them like a publish. `POST /flows/{id}/nodes/{node}/share` promotes a source to the library; -`/unshare` makes it private again. `GET /flows/library` lists shared sources. +`/unshare` makes it private again. `GET /flows/library` lists shared sources +and the `.` of every node using each. `DELETE /flows/library/{name}` retires one, and answers 409 while nodes still use it. With `?detach=true` each of those nodes keeps a copy of the code, in the published flow and the draft, and the shared source is deleted. @@ -149,8 +156,8 @@ values. You can publish only to declared messages, just as dashboard controls do | Method | Path | Action | |---|---|---| -| `POST` | `/runs/flows/{name}` | queue a run with `{"params": {...}, "seed": 7}` | -| `POST` | `/runs/flows/{name}/sweep` | queue runs with a shared group | +| `POST` | `/runs/flows/{flow id}` | queue a run with `{"params": {...}, "seed": 7}` | +| `POST` | `/runs/flows/{flow id}/sweep` | queue runs with a shared group | | `POST` | `/runs/import` | import up to 200 completed runs | | `GET` | `/runs` | query history | | `GET` | `/runs/overview` | summarize runs per flow | @@ -158,7 +165,7 @@ values. You can publish only to declared messages, just as dashboard controls do | `GET` | `/runs/{id}/nodes/{node}/outputs` | read what one node returned | | `POST` | `/runs/{id}/cancel`, `/runs/{id}/retry` | cancel, or create a retry | | `DELETE` | `/runs/{id}` | delete a finished run and its records | -| `DELETE` | `/runs?flow={name}` | delete a flow's runs; refused while any are active. `history=true` also clears its cascades from the history | +| `DELETE` | `/runs?flow={flow id}` | delete a flow's runs; refused while any are active. `history=true` also clears its cascades from the history | | `GET` | `/runs/{id}/metrics?name=&stride=` | read metric series | | `GET` | `/runs/metrics/names` | list metric names across a selection | | `GET` | `/runs/series/compare?ids=a,b,c&metric=&x=` | compare curves by step, time, or another metric | @@ -172,19 +179,29 @@ History filters include `flow`, `status`, `group`, `digest`, `since`, `before`, `limit`, and `offset`. Rows arrive newest first; use the last row's `created_at` as `before` for stable pagination. +History is kept by flow id, so it stays with a flow whose key changes and +stays apart from a later flow that takes a deleted one's key. In a run, a +cascade, an engine event and a metrics rollup, `flow` is the flow's id and +`flow_name` its key, or the key it last had once it is deleted. A `flow` +filter, here and under `/observability`, takes the id and only the id. Nodes +are shown and taken as `.`, and metric and artifact names keep the +key too (`train.loss`). + A node's outputs come from the stage cache, so only a node that succeeded and may be cached has them, up to 256 kB of JSON. Any other node answers `404`. Exports also accept `ids` and `until`, with `format=csv` or `jsonl`. -The metrics export is a long table; the runs export is one row per run. +The metrics export is a long table; the runs export is one row per run, with +the flow's id in `flow` and its key in `flow_name`. Use `params` and `metrics` to choose columns, including dotted paths into record outputs. An export covers at most the newest 10 000 runs. When a selection matches more, the response carries `X-Truncated` with that cap and `X-Total-Count` with the number of runs the selection matched. Import takes `{"runs": [...]}`, records completed results without executing -code, and deduplicates by `external_id`. See +code, and deduplicates by `external_id`. Each run's `flow` is the flow's id, +and its metric names are qualified with the key. See [Runs](../concepts/runs.md) for examples, artifacts, caching, and exports. ## Artifacts @@ -281,6 +298,12 @@ Connector state keys can also be listed and cleared by a superuser; see Changing flavors requires a superuser. See [Remote workers](workers.md). +The history under `/observability` is kept by flow id as described under +[Runs](#runs): `flow` is the id, `flow_name` the key, and `node` a +`.`, which is also how `timeseries` takes its `node` filter. Dead +letters are work the engine gave up on and name the flow and node by key, as +the engine runs them. + ## Live events Connect to `ws://127.0.0.1:8000/api/v1/flows/ws` with an @@ -289,6 +312,10 @@ Connect to `ws://127.0.0.1:8000/api/v1/flows/ws` with an Browsers first call `POST /flows/ws-ticket`, then connect with `?ticket=`. Tickets are single-use and expire after 30 seconds. The socket sends a snapshot followed by node, value, health, and run events. +An event about a flow names it by key in `flow`, the way its nodes and +messages are named, and by id in `flow_id`, which is empty for the engine as +a whole. `dashboard_changed` names the dashboard by id, or nothing when a +device's own wiring moved. A client asks for the bytes of the media it shows with `{"type": "media", "names": [...]}`, which replaces what it asked for before, diff --git a/docs/concepts/runs.md b/docs/concepts/runs.md index 6580940..1ee538d 100644 --- a/docs/concepts/runs.md +++ b/docs/concepts/runs.md @@ -42,7 +42,9 @@ curl -X POST "$FLUKSIO/runs/flows/train" \ ``` Here `$FLUKSIO` is the API base URL ending in `/api/v1`, and `$TOKEN` -is an access token. See [The HTTP API](../code/api.md) for authentication. +is an access token. The path takes the flow's id, here a flow whose id is +its key, as for one made before ids existed. See +[The HTTP API](../code/api.md) for authentication. Submission returns immediately with status `queued`. Read `GET /api/v1/runs/{id}` for status and results. Unknown parameters or wrong @@ -68,7 +70,7 @@ These four runs share a `group_id` and execute concurrently within the engine's limits. Each has its own state, so one run cannot overwrite another's messages. Vary seeds with `--param seed=1,2,3` if the flow declares that input. -The API equivalent is `POST /api/v1/runs/flows/{name}/sweep` with a +The API equivalent is `POST /api/v1/runs/flows/{flow id}/sweep` with a `runs` list of parameter dictionaries and optional seeds. Filter history with `GET /api/v1/runs?group=`. @@ -210,7 +212,11 @@ fluksio export runs --flow train --status ok -o results.csv Both include run IDs. Result rows also include input values, final numeric outputs, status, duration and code identifiers. Nested values use dotted paths, -such as `metric.final_metrics.train_loss`. +such as `metric.final_metrics.train_loss`. A row names the flow by its id in +`flow` and by its key in `flow_name`. History stays with the id, so a flow +keeps its runs whatever its key is later, and a new flow that takes the key +of a deleted one starts without them. Metric names keep the key, as in +`train.loss`. Python returns the same rows directly: @@ -243,8 +249,9 @@ Client().import_runs([{ }]) ``` -The flow must exist, and parameters are checked against its inputs. Repeated -imports of the same `external_id` reuse the existing record. Imported runs +The flow must exist, and parameters are checked against its inputs. Over the +HTTP API an entry's `flow` is the flow's id. Repeated imports of the same +`external_id` reuse the existing record. Imported runs have no code commit or digest because the engine did not execute them. Entries may also carry metric points and base64-encoded artifacts; see the API schema for their fields. @@ -257,7 +264,7 @@ the first run, and controls are disabled. If the flow has no dashboard, the same menu can create one from its streaming ports and outputs. The API equivalent is -`POST /api/v1/dashboards/from-flow/{name}`. The generated dashboard is +`POST /api/v1/dashboards/from-flow/{flow id}`. The generated dashboard is published and can be edited like any other dashboard. To keep a chart showing recent experiments, set its source to **Runs** and diff --git a/frontend/src/client/schemas.gen.ts b/frontend/src/client/schemas.gen.ts index 8ab2581..b778fd8 100644 --- a/frontend/src/client/schemas.gen.ts +++ b/frontend/src/client/schemas.gen.ts @@ -453,6 +453,13 @@ export const BrainNodeSchema = { type: 'array', title: 'Flows' }, + flow_ids: { + items: { + type: 'string' + }, + type: 'array', + title: 'Flow Ids' + }, issue: { anyOf: [ { @@ -970,7 +977,8 @@ export const DeadLetterSchema = { }, type: 'object', required: ['id', 'ts', 'flow', 'node', 'cause', 'reason'], - title: 'DeadLetter' + title: 'DeadLetter', + description: 'A work item given up on, named as the engine runs it: by key.' } as const; export const DeviceBindingSchema = { @@ -979,6 +987,11 @@ export const DeviceBindingSchema = { type: 'string', title: 'Flow' }, + flow_id: { + type: 'string', + title: 'Flow Id', + default: '' + }, node: { type: 'string', title: 'Node' @@ -1281,6 +1294,11 @@ export const EndpointSchema = { type: 'string', title: 'Id' }, + owner_id: { + type: 'string', + title: 'Owner Id', + default: '' + }, label: { type: 'string', title: 'Label' @@ -1357,6 +1375,11 @@ export const EventRowSchema = { type: 'string', title: 'Flow' }, + flow_name: { + type: 'string', + title: 'Flow Name', + default: '' + }, node: { type: 'string', title: 'Node' @@ -1891,6 +1914,11 @@ export const FlowRollupSchema = { type: 'string', title: 'Flow' }, + flow_name: { + type: 'string', + title: 'Flow Name', + default: '' + }, executions: { type: 'integer', title: 'Executions' @@ -1941,6 +1969,11 @@ export const FlowRunsRowSchema = { type: 'string', title: 'Flow' }, + flow_name: { + type: 'string', + title: 'Flow Name', + default: '' + }, runs: { type: 'integer', title: 'Runs' @@ -3985,6 +4018,11 @@ export const RunDetailSchema = { type: 'string', title: 'Flow' }, + flow_name: { + type: 'string', + title: 'Flow Name', + default: '' + }, status: { type: 'string', title: 'Status' @@ -4243,6 +4281,11 @@ export const SearchEntrySchema = { title: 'Parent', default: '' }, + id: { + type: 'string', + title: 'Id', + default: '' + }, kind: { type: 'string', title: 'Kind', @@ -5531,6 +5574,11 @@ export const fluksio__api__routes__observability__RunRowSchema = { type: 'string', title: 'Flow' }, + flow_name: { + type: 'string', + title: 'Flow Name', + default: '' + }, source: { type: 'string', title: 'Source' @@ -5593,6 +5641,11 @@ export const fluksio__api__routes__runs__RunRowSchema = { type: 'string', title: 'Flow' }, + flow_name: { + type: 'string', + title: 'Flow Name', + default: '' + }, status: { type: 'string', title: 'Status' diff --git a/frontend/src/client/sdk.gen.ts b/frontend/src/client/sdk.gen.ts index 95e5484..6637662 100644 --- a/frontend/src/client/sdk.gen.ts +++ b/frontend/src/client/sdk.gen.ts @@ -1902,6 +1902,9 @@ export class FlowsService { /** * Read Training * A wren node's latest training run, or null when it has none. + * + * Found by the node it trains alone: only a training run names one, and + * the training flow may have been registered again under another id since. * @param data The data for the request. * @param data.flowId * @param data.nodeId @@ -2370,6 +2373,8 @@ export class ObservabilityService { * the same pair ``/runs`` and ``/events`` take — so a chart dragged to a * span can be re-fetched at that span's own resolution rather than * magnifying the buckets it already has. + * + * ``flow`` is a flow id, ``node`` a ``.`` as the engine names it. * @param data The data for the request. * @param data.flow * @param data.node @@ -2449,6 +2454,7 @@ export class ObservabilityService { * * ``since`` is inclusive and ``until`` exclusive, so a window of one minute * holds exactly the runs of the minute bucket the charts are drawn from. + * ``flow`` is a flow id. * @param data The data for the request. * @param data.flow * @param data.status @@ -2483,7 +2489,8 @@ export class ObservabilityService { * takes, so a list can cover the span the charts beside it are drawn from. * ``run`` narrows to one run — a batch run's id, or the journaled item a live * cascade came from. Rows recorded before the column existed carry none, so - * an old failure answers no run at all rather than the wrong one. + * an old failure answers no run at all rather than the wrong one. ``flow`` + * is a flow id. * @param data The data for the request. * @param data.kind * @param data.flow @@ -2577,7 +2584,7 @@ export class RunsService { * Create Run * Queue one run of a flow. * @param data The data for the request. - * @param data.name + * @param data.flowId * @param data.requestBody * @returns fluksio__api__routes__runs__RunRow Successful Response * @throws ApiError @@ -2585,9 +2592,9 @@ export class RunsService { public static createRun(data: RunsCreateRunData): CancelablePromise { return __request(OpenAPI, { method: 'POST', - url: '/api/v1/runs/flows/{name}', + url: '/api/v1/runs/flows/{flow_id}', path: { - name: data.name + flow_id: data.flowId }, body: data.requestBody, mediaType: 'application/json', @@ -2605,7 +2612,7 @@ export class RunsService { * search is this with the parameters spread out. Either way the caller * builds the list — the engine does not own a sweep grammar. * @param data The data for the request. - * @param data.name + * @param data.flowId * @param data.requestBody * @returns fluksio__api__routes__runs__RunRow Successful Response * @throws ApiError @@ -2613,9 +2620,9 @@ export class RunsService { public static createSweep(data: RunsCreateSweepData): CancelablePromise { return __request(OpenAPI, { method: 'POST', - url: '/api/v1/runs/flows/{name}/sweep', + url: '/api/v1/runs/flows/{flow_id}/sweep', path: { - name: data.name + flow_id: data.flowId }, body: data.requestBody, mediaType: 'application/json', @@ -2658,7 +2665,7 @@ export class RunsService { * first, so handing back the last row's ``created_at`` reads the next page * whatever landed meanwhile — which ``offset`` cannot, since a run submitted * between two pages shifts every row down one. ``since`` bounds the other - * end and is inclusive. + * end and is inclusive. ``flow`` is a flow id. * @param data The data for the request. * @param data.flow * @param data.status @@ -2693,7 +2700,7 @@ export class RunsService { /** * Delete Runs - * Every run of one flow at once. + * Every run of one flow at once, ``flow`` being its id. * * The counterpart to the list's ``flow`` filter, and what a reseed needs: * deleting a flow leaves its runs standing on purpose, so this is the only @@ -2915,8 +2922,9 @@ export class RunsService { * Read Node Outputs * What one node returned, by message, as the stage cache stored it. * - * Kept only for a node that succeeded and may be reused, and only up to - * 256 kB of JSON. Null when the node published nothing. + * The node is ``.``, as the run shows it. Kept only for a node + * that succeeded and may be reused, and only up to 256 kB of JSON. Null when + * the node published nothing. * @param data The data for the request. * @param data.runId * @param data.node diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts index 231fac4..6423580 100644 --- a/frontend/src/client/types.gen.ts +++ b/frontend/src/client/types.gen.ts @@ -112,6 +112,7 @@ export type BrainNode = { kind: string; members?: Array<(string)>; flows?: Array<(string)>; + flow_ids?: Array<(string)>; /** * Why this neuron cannot run, if validation found something. A failure the engine hits while running arrives over the socket instead; this is the part that is already true before anything fires, and so has to travel with the graph. */ @@ -241,6 +242,9 @@ export type DashboardSummary = { updated_at?: (string | null); }; +/** + * A work item given up on, named as the engine runs it: by key. + */ export type DeadLetter = { id: string; ts: number; @@ -255,6 +259,7 @@ export type DeadLetter = { */ export type DeviceBinding = { flow: string; + flow_id?: string; node: string; port: string; message: string; @@ -380,6 +385,7 @@ export type DType = 'float' | 'int' | 'str' | 'bool' | 'json' | 'series' | 'reco export type Endpoint = { kind: string; id: string; + owner_id?: string; label: string; detail?: string; provides?: Array<(string)>; @@ -396,6 +402,7 @@ export type EventRow = { ts: string; type: string; flow: string; + flow_name?: string; node: string; detail: string; actor: string; @@ -569,6 +576,7 @@ export type FlowOrigin = { export type FlowRollup = { flow: string; + flow_name?: string; executions: number; errors: number; messages: number; @@ -583,6 +591,7 @@ export type FlowRollup = { */ export type FlowRunsRow = { flow: string; + flow_name?: string; runs: number; running: number; queued: number; @@ -641,6 +650,7 @@ export type fluksio__api__routes__messages__PublishRequest = { export type fluksio__api__routes__observability__RunRow = { id: string; flow: string; + flow_name?: string; source: string; parent?: string; status: string; @@ -658,6 +668,7 @@ export type fluksio__api__routes__observability__RunRow = { export type fluksio__api__routes__runs__RunRow = { id: string; flow: string; + flow_name?: string; status: string; status_reason: string; cause: string; @@ -1305,6 +1316,7 @@ export type cause = 'api' | 'cli' | 'sdk'; export type RunDetail = { id: string; flow: string; + flow_name?: string; status: string; status_reason: string; cause: string; @@ -1373,6 +1385,7 @@ export type SearchEntry = { name: string; title?: string; parent?: string; + id?: string; kind?: string; detail?: string; }; @@ -2352,14 +2365,14 @@ export type PrivateCreateUserData = { export type PrivateCreateUserResponse = (UserPublic); export type RunsCreateRunData = { - name: string; + flowId: string; requestBody: RunCreate; }; export type RunsCreateRunResponse = (fluksio__api__routes__runs__RunRow); export type RunsCreateSweepData = { - name: string; + flowId: string; requestBody: SweepCreate; }; diff --git a/frontend/src/components/Runs/NewRun.tsx b/frontend/src/components/Runs/NewRun.tsx index c2e03d2..4e7394a 100644 --- a/frontend/src/components/Runs/NewRun.tsx +++ b/frontend/src/components/Runs/NewRun.tsx @@ -178,7 +178,7 @@ function RunForm({ flow }: { flow: string }) { mutationFn: async () => { if (grid.length > 1) { return RunsService.createSweep({ - name: flow, + flowId: flow, requestBody: { ...flags, runs: grid.map((params) => ({ @@ -190,7 +190,7 @@ function RunForm({ flow }: { flow: string }) { }) } return RunsService.createRun({ - name: flow, + flowId: flow, requestBody: { ...flags, ...withSeed,