stroblmeandClaude Opus 5
180da3d640
The engine was I/O-bound on its own state backend. `RedisState.lock()` is one key — `pipeline:_lock` — for the whole process, taken five times a message at two round trips each, and every cascade and every node read queued behind it. Inside it, reading a node's inputs was three round trips per input (an EXISTS for `in`, then EXISTS and GET for the value), writing was two updates that a single transaction already gives, and the version counters went one INCR at a time. Replaced with the atomic command that was always available: `get_present` is one MGET and tells a missing key from one holding null, so the lock it used to be read under bought nothing; value and timestamp land in one `update`, which is a MULTI/EXEC; `increment_multi` pipelines the counters. `values()` — what every websocket snapshot calls — is two reads whatever the message count instead of two per message. Beside that: every webhook did its blocking XADD on the asyncio event loop (MQTT already used `to_thread`); the per-execution `NodeOutcome` was built and validated even with no run watching; `_minute` built a tz-aware datetime per event on the loop thread to key a dict, and now keys on an int; `move_due` promoted delayed items one round trip each, every second; `FLOW_MAX_CASCADES` makes the in-flight ceiling a setting rather than a constant. `orjson` replaces stdlib json where a message pays for it — state, the journal, the engine side of the worker pipe. `fluksio-worker` stays dependency-free, and the run-cache digest stays on stdlib so no stored key is invalidated. A non-finite number now stores as `null` rather than the bare `NaN` that was never JSON. Measured with `scripts/bench_engine.py` against a real Redis, 200 messages: a five-node chain went from 43.9 to 103.1 msg/s with p50 latency 2110ms → 782ms and p95 3913ms → 1439ms; one source into twenty consumers went from 5.4 to 33.7 msg/s. In memory, twenty consumers went from 187 to 448 msg/s. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01BpfSinyCBfjuieikyfMPbf
411 lines
16 KiB
Python
411 lines
16 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
|
|
#: ``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())
|
|
run = self._runs.get(str(event.get("run") or ""))
|
|
|
|
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
|
|
],
|
|
)
|
|
)
|
|
return
|
|
|
|
if kind == "cascade_started":
|
|
self._start_run(event, ts)
|
|
return
|
|
|
|
if kind == "cascade_finished":
|
|
if run is not None:
|
|
run["finished_at"] = datetime.fromtimestamp(ts, UTC)
|
|
run["duration_ms"] = round((ts - run["started_ts"]) * 1000, 2)
|
|
run["status"] = "error" if run["errors"] else "ok"
|
|
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.",
|
|
)
|
|
)
|
|
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),
|
|
)
|
|
)
|
|
|
|
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" or 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 run still open long after it started did not finish; saying so is
|
|
# more honest than leaving it running forever.
|
|
session.execute(
|
|
update(FlowRun)
|
|
.where(
|
|
col(FlowRun.status) == "running",
|
|
col(FlowRun.started_at) < now - timedelta(seconds=RUN_STALE_S),
|
|
)
|
|
.values(status="abandoned", finished_at=now)
|
|
)
|