Fluksio/app
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

History is kept by flow id: runs, cascades, rollups and events file the id in their flow column and nodes as <flow id>.<node>, the API answers flow plus flow_name and shows and takes nodes as <key>.<node>; events carry flow_id from the moment they are published; run queue items, the stage cache and cached curves keep ids and keys apart; endpoints, the brain graph, device bindings and search name the id to open

Browse Source
This commit is contained in:
stroblme
2026-11-02 19:01:19 +01:00
parent e6027af8fe
commit d4a1b1db47
34 changed files with 1374 additions and 329 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/fluksio/api/routes/flavors.py
+6 -5
View File
@@ -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
``<key>.<node>``."""
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
backend/fluksio/api/routes/flows.py
+42 -22
View File
@@ -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:
backend/fluksio/api/routes/observability.py
+59 -14
View File
@@ -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 = ""
#: ``<key>.<node>``, 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 ``<key>.<node>`` 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 `<flow id>.<node>`, 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])
backend/fluksio/api/routes/runs.py
+164 -69
View File
@@ -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):
#: ``<key>.<node>``, the way the engine names it.
node: str
status: str
attempt: int
@@ -153,6 +161,7 @@ class RunNodeRow(BaseModel):
class ArtifactRow(BaseModel):
name: str
#: ``<key>.<node>``.
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 `<flow id>.<node>`, 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 ``<key>.<node>``, 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)
backend/fluksio/api/routes/search.py
+20 -8
View File
@@ -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,
#: ``<node>.<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, ``<node>.<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
backend/fluksio/flow/controller.py
+15 -12
View File
@@ -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:
backend/fluksio/flow/dashboards.py
+5 -3
View File
@@ -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,
backend/fluksio/flow/devices.py
+13 -5
View File
@@ -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]:
backend/fluksio/flow/events.py
+11 -1
View File
@@ -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.
backend/fluksio/flow/metrics.py
+30 -16
View File
@@ -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 ``<flow id>.<node>`` 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,
backend/fluksio/flow/runs.py
+69 -33
View File
@@ -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: ``<flow id>.<node>``.
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
``<flow id>.<node>``.
"""
# 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 ``<flow id>.<node>``. 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 "",
backend/fluksio/flow/schemas.py
+3
View File
@@ -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=(
backend/fluksio/flow/store.py
+19 -12
View File
@@ -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 ``<key>.<node>`` 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."""
backend/fluksio/flow/tombstones.py
+47 -5
View File
@@ -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 ``<flow id>.<node>``. The API shows and takes it as
``<key>.<node>``, 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 ``<flow id>.<node>`` as ``<key>.<node>``, 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 ``<key>.<node>`` as history stores it, ``<flow id>.<node>``.
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 = [
backend/fluksio/main.py
+4
View File
@@ -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
backend/fluksio/models.py
+15 -7
View File
@@ -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))
#: ``<flow id>.<node>``.
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))
#: ``<flow id>.<node>``, 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 ``<flow id>.<node>``: 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
backend/tests/api/routes/test_devices.py
+6 -3
View File
@@ -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
backend/tests/api/routes/test_flow_ids.py
+62 -2
View File
@@ -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 ``<key>.<node>``, 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()
backend/tests/api/routes/test_flows.py
+18 -8
View File
@@ -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(
backend/tests/api/routes/test_history_ids.py
+498
View File
@@ -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 = "<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),
},
# `<event type>:<node or flow key>`, 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")
backend/tests/api/routes/test_observability.py
+28 -20
View File
@@ -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,
backend/tests/api/routes/test_runs.py
+51 -36
View File
@@ -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),
)
backend/tests/api/routes/test_search.py
+14 -2
View File
@@ -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(
backend/tests/api/routes/test_wren_train.py
+17 -4
View File
@@ -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},
backend/tests/test_artifacts.py
+1 -1
View File
@@ -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(
backend/tests/test_db.py
+4 -2
View File
@@ -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
backend/tests/test_metrics.py
+13 -7
View File
@@ -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")
backend/tests/test_tombstones.py
+4 -4
View File
@@ -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"
docs/code/api.md
+33 -6
View File
@@ -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
`<key>.<node>` 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 `<key>.<node>` 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 `<key>.<node>`, 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
`<key>.<node>`, 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=<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,
docs/concepts/runs.md
+13 -6
View File
@@ -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=<group-id>`.
@@ -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
frontend/src/client/schemas.gen.ts
+54 -1
View File
@@ -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'
frontend/src/client/sdk.gen.ts
+19 -11
View File
@@ -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 ``<key>.<node>`` 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<RunsCreateRunResponse> {
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<RunsCreateSweepResponse> {
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 ``<key>.<node>``, 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
frontend/src/client/types.gen.ts
+15 -2
View File
@@ -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;
};
frontend/src/components/Runs/NewRun.tsx
+2 -2
View File
@@ -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,