499 lines
17 KiB
Python
499 lines
17 KiB
Python
"""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")
|