stroblmeandClaude Opus 5
4479eeb726
Docs / docs (push) Successful in 38s
Playwright Tests / test-playwright (1, 2) (push) Successful in 2m56s
Playwright Tests / test-playwright (2, 2) (push) Successful in 2m7s
pre-commit / pre-commit (push) Failing after 2m17s
Test Backend / test-backend (push) Successful in 2m54s
Compose Smoke Test / test-compose (push) Successful in 44s
Playwright Tests / merge-reports (push) Successful in 1m17s
Three things the first export pass got wrong for a real study. **Dotted paths.** A node returns a record, not a scalar — the numbers arrive inside `final_metrics` — so `--metrics final_metrics.train_loss` yielded an empty column and `--metrics final_metrics` yielded the whole record in one cell. Both sides of the wide table now take dotted paths, and the defaults reach the same depth: every number a result carries is a column named by its path, and inputs are compared leaf by leaf, so two configurations differing in one field give that field as the axis rather than two blobs that are merely not equal. Lists stay whole — a curve belongs in the long table. **`--list`.** Metric names are flow-qualified, so `--name train_loss` matched nothing and said only that. `fluksio export metrics --list` prints the names the selection carries, and an empty export made with `--name` points at it. **A version to compare.** The CLI ships ahead of the engine and a stale one answered a flat 404 with nothing anywhere in the API to tell how old it was. The engine reports `version` on `/observability/summary`, `fluksio status` prints it, and a 404 from export now names both versions — or says "older" when the field itself predates the engine. Bumped to 0.1.5, which is what makes the number worth reading. Also formats `flow/metrics.py`, which had been committed unformatted and was the last `ruff format --check` failure. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01A9Hdrmf2cwNABCnE5x9UJa
444 lines
18 KiB
Python
444 lines
18 KiB
Python
"""What the engine did, kept long enough to answer for it.
|
|
|
|
The event bus already carries every execution, error and cascade; until now
|
|
nothing wrote any of it down, so "was it slow yesterday?" had no answer. This
|
|
subscriber folds those events into per-minute rollups, keeps the failures and
|
|
the audit trail whole, and records one row per cascade.
|
|
|
|
Accumulation is in memory and flushed every few seconds: a node firing at
|
|
10 Hz must not be 10 inserts a second, and the arithmetic that turns it into
|
|
one row a minute is cheaper than the round trip would be.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import time
|
|
from datetime import UTC, datetime, timedelta
|
|
from typing import Any
|
|
|
|
from sqlalchemy import delete, func, update
|
|
from sqlalchemy.dialects.sqlite import insert
|
|
from sqlmodel import Session, col
|
|
|
|
from fluksio.core.config import settings
|
|
from fluksio.core.db import engine
|
|
from fluksio.flow.events import EventBus
|
|
from fluksio.models import EngineEvent, FlowRun, MetricBucket
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
#: How often the accumulated minute is written out.
|
|
FLUSH_INTERVAL_S = 15.0
|
|
#: A traceback is worth reading; a whole run of a chatty node is not.
|
|
DETAIL_CAP = 8000
|
|
#: A cascade still open this long after it started is never finishing.
|
|
RUN_STALE_S = 600.0
|
|
#: What ``flow_run.source`` says for a batch run, as against the cause a
|
|
#: cascade carries. A training step can legitimately run for hours, so this is
|
|
#: also what exempts one from the staleness sweep above.
|
|
RUN_SOURCE = "run"
|
|
#: ``metric_minute.flow``/``node`` and ``flow_run.flow`` are this wide, and a
|
|
#: node id has no length of its own. One row over it aborts the transaction,
|
|
#: which would take every other row in the flush with it.
|
|
NAME_MAX = 255
|
|
#: How much unwritten history is held while the database is unreachable. Enough
|
|
#: for a long outage, bounded so the outage cannot become a memory leak.
|
|
HOLD_MAX = 10_000
|
|
#: Retention is checked this often, not on every flush.
|
|
PRUNE_INTERVAL_S = 3600.0
|
|
|
|
#: Bucket columns that add up over a minute, and the two that take the larger.
|
|
SUMMED = (
|
|
"executions",
|
|
"errors",
|
|
"messages",
|
|
"duration_sum_ms",
|
|
"lag_sum_ms",
|
|
"items",
|
|
)
|
|
MAXIMA = ("duration_max_ms", "lag_max_ms")
|
|
|
|
#: Engine events kept as rows. Everything else on the bus is traffic.
|
|
RECORDED = {
|
|
"flow_quarantined",
|
|
"task_crashed",
|
|
"engine_degraded",
|
|
"engine_fatal",
|
|
"cascade_dropped",
|
|
"queue_unavailable",
|
|
}
|
|
|
|
|
|
def _minute(ts: float) -> int:
|
|
"""The epoch second the minute containing ``ts`` starts at.
|
|
|
|
An integer rather than a datetime: this runs once per event on the API's
|
|
event loop, and building a tz-aware datetime to key a dict with cost more
|
|
than everything the collector does with the event afterwards. The row
|
|
still wants one, so `_write` builds it — once per minute per node rather
|
|
than once per event.
|
|
"""
|
|
return int(ts // 60) * 60
|
|
|
|
|
|
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":
|
|
text = f"Given up on after {event.get('deliveries')} deliveries. {text}"
|
|
return text[:DETAIL_CAP]
|
|
|
|
|
|
class MetricsCollector:
|
|
"""Folds engine events into rollups, failures and run records."""
|
|
|
|
def __init__(self, events: EventBus, flush_s: float = FLUSH_INTERVAL_S) -> None:
|
|
self._events = events
|
|
self._flush_s = flush_s
|
|
self._buckets: dict[tuple[str, str, int], dict[str, float]] = {}
|
|
self._runs: dict[str, dict[str, Any]] = {}
|
|
self._pending: list[EngineEvent] = []
|
|
# The traceback arrives one event before the failure it belongs to,
|
|
# held with the time it arrived so an unpaired one does not stay put.
|
|
self._tracebacks: dict[tuple[str, str], tuple[float, str]] = {}
|
|
self._last_prune = 0.0
|
|
|
|
# -------------------------------------------------------------------------
|
|
# The loop
|
|
# -------------------------------------------------------------------------
|
|
|
|
async def run(self) -> None:
|
|
"""Consume the bus until cancelled, flushing on a fixed interval."""
|
|
last = time.monotonic()
|
|
async with self._events.subscribe() as queue:
|
|
while True:
|
|
# A busy bus never idles, so the flush is on a deadline rather
|
|
# than on the timeout alone.
|
|
timeout = max(0.05, self._flush_s - (time.monotonic() - last))
|
|
try:
|
|
event = await asyncio.wait_for(queue.get(), timeout)
|
|
except TimeoutError:
|
|
pass
|
|
else:
|
|
try:
|
|
self.handle(event)
|
|
except Exception:
|
|
logger.exception("Could not record %s", event.get("type"))
|
|
if time.monotonic() - last >= self._flush_s:
|
|
await self.flush()
|
|
last = time.monotonic()
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Accumulating
|
|
# -------------------------------------------------------------------------
|
|
|
|
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())),
|
|
)
|
|
return self._buckets.setdefault(
|
|
key,
|
|
{
|
|
"executions": 0,
|
|
"errors": 0,
|
|
"messages": 0,
|
|
"duration_sum_ms": 0.0,
|
|
"duration_max_ms": 0.0,
|
|
"lag_sum_ms": 0.0,
|
|
"lag_max_ms": 0.0,
|
|
"items": 0,
|
|
},
|
|
)
|
|
|
|
def handle(self, event: dict[str, Any]) -> None:
|
|
"""Fold one event in. Synchronous: this is arithmetic on dicts."""
|
|
kind = str(event.get("type") or "")
|
|
ts = float(event.get("ts") or time.time())
|
|
# Two different things: the id as the event carries it, which is what a
|
|
# recorded row keeps, and the live cascade it belongs to — which a batch
|
|
# run has none of, since nothing here started one for it.
|
|
run_id = str(event.get("run") or "")[:64]
|
|
run = self._runs.get(run_id)
|
|
|
|
if kind == "node_executed":
|
|
bucket = self._bucket(event)
|
|
bucket["executions"] += 1
|
|
bucket["messages"] += int(event.get("outputs") or 0)
|
|
duration = float(event.get("duration_ms") or 0.0)
|
|
bucket["duration_sum_ms"] += duration
|
|
bucket["duration_max_ms"] = max(bucket["duration_max_ms"], duration)
|
|
if run is not None:
|
|
run["nodes"] += 1
|
|
return
|
|
|
|
if kind == "work_latency":
|
|
bucket = self._bucket(event)
|
|
lag = float(event.get("lag_ms") or 0.0)
|
|
bucket["lag_sum_ms"] += lag
|
|
bucket["lag_max_ms"] = max(bucket["lag_max_ms"], lag)
|
|
bucket["items"] += 1
|
|
return
|
|
|
|
if kind == "node_log":
|
|
# Held for the node_error that follows it from the same thread.
|
|
if event.get("level") == "error":
|
|
key = (str(event.get("flow") or ""), str(event.get("node") or ""))
|
|
self._tracebacks[key] = (ts, str(event.get("text") or "")[:DETAIL_CAP])
|
|
return
|
|
|
|
if kind == "node_error":
|
|
self._bucket(event)["errors"] += 1
|
|
if run is not None:
|
|
run["errors"] += 1
|
|
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 "")
|
|
self._pending.append(
|
|
EngineEvent(
|
|
ts=datetime.fromtimestamp(ts, UTC),
|
|
type="node_error",
|
|
flow=str(event.get("flow") or ""),
|
|
node=str(event.get("node") or ""),
|
|
detail=(f"{error}\n{traceback}" if traceback else error)[
|
|
:DETAIL_CAP
|
|
],
|
|
run=run_id,
|
|
)
|
|
)
|
|
return
|
|
|
|
if kind == "cascade_started":
|
|
self._start_run(event, ts)
|
|
return
|
|
|
|
if kind == "cascade_finished":
|
|
if run is not None:
|
|
self._finish_run(run, ts)
|
|
return
|
|
|
|
# A batch run opens no cascade, so without these it ran, recorded its
|
|
# nodes and its failures, and appeared in no history at all.
|
|
if kind == "run_started":
|
|
self._start_run({**event, "cause": RUN_SOURCE}, ts)
|
|
return
|
|
|
|
if kind == "run_finished":
|
|
if run is not None:
|
|
self._finish_run(run, ts, str(event.get("status") or ""))
|
|
return
|
|
|
|
if kind == "node_health":
|
|
if event.get("health") == "down":
|
|
self._pending.append(
|
|
EngineEvent(
|
|
ts=datetime.fromtimestamp(ts, UTC),
|
|
type="node_health",
|
|
flow=str(event.get("flow") or ""),
|
|
node=str(event.get("node") or ""),
|
|
detail=_detail(event) or "Reported itself down.",
|
|
run=run_id,
|
|
)
|
|
)
|
|
return
|
|
|
|
if kind == "audit":
|
|
self._pending.append(
|
|
EngineEvent(
|
|
ts=datetime.fromtimestamp(ts, UTC),
|
|
type="audit",
|
|
flow=str(event.get("flow") or ""),
|
|
detail=str(event.get("action") or ""),
|
|
actor=str(event.get("user") or ""),
|
|
)
|
|
)
|
|
return
|
|
|
|
if kind in RECORDED:
|
|
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 ""),
|
|
detail=_detail(event),
|
|
run=run_id,
|
|
)
|
|
)
|
|
|
|
def _finish_run(self, run: dict[str, Any], ts: float, status: str = "") -> None:
|
|
"""Close an open record. A run says how it ended; a cascade is told."""
|
|
run["finished_at"] = datetime.fromtimestamp(ts, UTC)
|
|
run["duration_ms"] = round((ts - run["started_ts"]) * 1000, 2)
|
|
run["status"] = status or ("error" if run["errors"] else "ok")
|
|
|
|
def _start_run(self, event: dict[str, Any], ts: float) -> None:
|
|
run_id = str(event.get("run") or "")
|
|
if not run_id:
|
|
return
|
|
existing = self._runs.get(run_id)
|
|
if existing is not None:
|
|
# A redelivery of the same item: one run, tried again.
|
|
existing["deliveries"] = int(event.get("deliveries") or 1)
|
|
existing["status"] = "running"
|
|
existing["finished_at"] = None
|
|
return
|
|
self._runs[run_id] = {
|
|
"id": run_id,
|
|
"flow": str(event.get("flow") or "")[:NAME_MAX],
|
|
"source": str(event.get("cause") or ""),
|
|
"started_ts": ts,
|
|
"started_at": datetime.fromtimestamp(ts, UTC),
|
|
"finished_at": None,
|
|
"status": "running",
|
|
"nodes": 0,
|
|
"errors": 0,
|
|
"duration_ms": 0.0,
|
|
"deliveries": int(event.get("deliveries") or 1),
|
|
}
|
|
|
|
# -------------------------------------------------------------------------
|
|
# Writing
|
|
# -------------------------------------------------------------------------
|
|
|
|
async def flush(self) -> None:
|
|
buckets, self._buckets = self._buckets, {}
|
|
pending, self._pending = self._pending, []
|
|
# Open runs stay in memory: their counts are still growing, and the row
|
|
# is written from the whole record each time rather than in deltas.
|
|
runs = list(self._runs.values())
|
|
prune = time.monotonic() - self._last_prune >= PRUNE_INTERVAL_S
|
|
if not (buckets or pending or runs or prune):
|
|
return
|
|
try:
|
|
await asyncio.to_thread(self._write, buckets, pending, runs, prune)
|
|
except Exception:
|
|
logger.exception("Could not write engine metrics")
|
|
self._hold(buckets, pending)
|
|
return
|
|
if prune:
|
|
self._last_prune = time.monotonic()
|
|
cutoff = time.time() - RUN_STALE_S
|
|
for run_id, run in list(self._runs.items()):
|
|
if run["status"] != "running":
|
|
del self._runs[run_id]
|
|
# A batch run is held however long it takes: a training step of an
|
|
# hour is a normal one, and `run_finished` is what ends it.
|
|
elif run["source"] != RUN_SOURCE and run["started_ts"] < cutoff:
|
|
del self._runs[run_id]
|
|
# Held tracebacks survive the flush: the log and the failure it belongs
|
|
# to are two events, and a flush can fall between them. One per node,
|
|
# each replaced by that node's next failure — and dropped once it is
|
|
# clear the failure it was waiting for is never coming.
|
|
for key, (held, _) in list(self._tracebacks.items()):
|
|
if held < cutoff:
|
|
del self._tracebacks[key]
|
|
|
|
def _hold(
|
|
self,
|
|
buckets: dict[tuple[str, str, int], dict[str, float]],
|
|
pending: list[EngineEvent],
|
|
) -> None:
|
|
"""Take a batch the database refused back, rather than losing it.
|
|
|
|
These are the audit trail and the failures with their tracebacks — the
|
|
rows nobody can reconstruct afterwards. The batch goes in front of what
|
|
arrived since, so what a long outage drops is the oldest.
|
|
"""
|
|
for key, agg in self._buckets.items():
|
|
held = buckets.get(key)
|
|
if held is None:
|
|
buckets[key] = agg
|
|
continue
|
|
for name in SUMMED:
|
|
held[name] += agg[name]
|
|
for name in MAXIMA:
|
|
held[name] = max(held[name], agg[name])
|
|
self._buckets = buckets
|
|
# A failed commit puts the rows back to unsaved. One that got as far as
|
|
# being handed an id keeps it, and the sequence has already moved past
|
|
# it, so adding them to the next session inserts them cleanly.
|
|
self._pending[:0] = pending
|
|
|
|
for key in list(self._buckets)[: max(0, len(self._buckets) - HOLD_MAX)]:
|
|
del self._buckets[key]
|
|
del self._pending[: max(0, len(self._pending) - HOLD_MAX)]
|
|
|
|
def _write(
|
|
self,
|
|
buckets: dict[tuple[str, str, int], dict[str, float]],
|
|
pending: list[EngineEvent],
|
|
runs: list[dict[str, Any]],
|
|
prune: bool,
|
|
) -> None:
|
|
with Session(engine) as session:
|
|
for (flow, node, minute), agg in buckets.items():
|
|
statement = insert(MetricBucket).values(
|
|
flow=flow,
|
|
node=node,
|
|
bucket=datetime.fromtimestamp(minute, UTC),
|
|
**agg,
|
|
)
|
|
# The same minute is written several times, so the counters add
|
|
# and the maxima take whichever is larger. Columns are read by
|
|
# subscript: `excluded.items` is the collection's own method.
|
|
new = statement.excluded
|
|
session.execute(
|
|
statement.on_conflict_do_update(
|
|
index_elements=["flow", "node", "bucket"],
|
|
set_={
|
|
name: col(getattr(MetricBucket, name)) + new[name]
|
|
for name in SUMMED
|
|
}
|
|
| {
|
|
# `max` of two values, not the aggregate: SQLite's
|
|
# scalar form, which is what `greatest` is elsewhere.
|
|
name: func.max(col(getattr(MetricBucket, name)), new[name])
|
|
for name in MAXIMA
|
|
},
|
|
)
|
|
)
|
|
|
|
for run in runs:
|
|
values = {k: v for k, v in run.items() if k != "started_ts"}
|
|
statement = insert(FlowRun).values(**values)
|
|
session.execute(
|
|
statement.on_conflict_do_update(
|
|
index_elements=["id"],
|
|
set_={
|
|
key: statement.excluded[key]
|
|
for key in values
|
|
if key != "id"
|
|
},
|
|
)
|
|
)
|
|
|
|
session.add_all(pending)
|
|
if prune:
|
|
self._prune(session)
|
|
session.commit()
|
|
|
|
def _prune(self, session: Session) -> None:
|
|
now = datetime.now(UTC)
|
|
cutoff = now - timedelta(days=settings.OBS_RETENTION_DAYS)
|
|
session.execute(delete(MetricBucket).where(col(MetricBucket.bucket) < cutoff))
|
|
session.execute(delete(EngineEvent).where(col(EngineEvent.ts) < cutoff))
|
|
session.execute(delete(FlowRun).where(col(FlowRun.started_at) < cutoff))
|
|
# A cascade still open long after it started did not finish; saying so
|
|
# is more honest than leaving it running forever. A batch run is not
|
|
# swept: it is allowed to take hours, and `Run` has its own lease sweep
|
|
# for the case where the engine died holding one.
|
|
session.execute(
|
|
update(FlowRun)
|
|
.where(
|
|
col(FlowRun.status) == "running",
|
|
col(FlowRun.source) != RUN_SOURCE,
|
|
col(FlowRun.started_at) < now - timedelta(seconds=RUN_STALE_S),
|
|
)
|
|
.values(status="abandoned", finished_at=now)
|
|
)
|