stroblmeandClaude Fable 5
0b91379914
/observability/runs gains since/until, so the throughput chart's pin asks the server for its minute instead of filtering a fixed recent list. This engine writes ~60 runs a minute, so any minute but the newest read empty. since is inclusive and until exclusive, matching the minute buckets the charts are drawn from. Hover stays the client-side preview it was: scrubbing a day would otherwise be a request per minute rested on. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_017MeiWk3Yq12n2pTvnQWYvt
328 lines
10 KiB
Python
328 lines
10 KiB
Python
"""What the engine has been doing: health now, and history since.
|
|
|
|
The live state comes from the controller; everything older than the websocket's
|
|
memory comes from the rollups the metrics collector writes. Deliberately not
|
|
built on ``/utils/health/``: that endpoint answers 503 when something is wrong,
|
|
which the generated SDK turns into a thrown error — and a health page that
|
|
cannot render while the engine is degraded is the wrong way round.
|
|
"""
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Any, Literal
|
|
|
|
from fastapi import APIRouter, Depends, Request
|
|
from fastapi.concurrency import run_in_threadpool
|
|
from pydantic import BaseModel
|
|
from sqlalchemy import func
|
|
from sqlmodel import col, select
|
|
|
|
from app.api.deps import FlowControllerDep, SessionDep, get_current_user
|
|
from app.flow.controller import NodeStatus
|
|
from app.models import EngineEvent, FlowRun, MetricBucket
|
|
|
|
router = APIRouter(
|
|
prefix="/observability",
|
|
tags=["observability"],
|
|
dependencies=[Depends(get_current_user)],
|
|
)
|
|
|
|
#: How many slices a per-flow sparkline is folded into.
|
|
SPARK_SLICES = 60
|
|
|
|
|
|
class HealthSummary(BaseModel):
|
|
status: str
|
|
problems: list[str]
|
|
flows: dict[str, int]
|
|
nodes: dict[str, int]
|
|
queue: dict[str, Any]
|
|
loop_lag: dict[str, float]
|
|
failures_24h: int
|
|
|
|
|
|
class SeriesPoint(BaseModel):
|
|
ts: float
|
|
executions: int
|
|
errors: int
|
|
messages: int
|
|
avg_ms: float
|
|
max_ms: float
|
|
avg_lag_ms: float
|
|
|
|
|
|
class FlowRollup(BaseModel):
|
|
flow: str
|
|
executions: int
|
|
errors: int
|
|
messages: int
|
|
avg_ms: float
|
|
avg_lag_ms: float
|
|
spark: list[int]
|
|
last_error_ts: float | None = None
|
|
|
|
|
|
class RunRow(BaseModel):
|
|
id: str
|
|
flow: str
|
|
source: str
|
|
status: str
|
|
started_at: datetime
|
|
finished_at: datetime | None = None
|
|
nodes: int
|
|
errors: int
|
|
duration_ms: float
|
|
deliveries: int
|
|
|
|
|
|
class EventRow(BaseModel):
|
|
id: int
|
|
ts: datetime
|
|
type: str
|
|
flow: str
|
|
node: str
|
|
detail: str
|
|
actor: str
|
|
|
|
|
|
class DeadLetter(BaseModel):
|
|
id: str
|
|
ts: float
|
|
flow: str
|
|
node: str
|
|
cause: str
|
|
reason: str
|
|
|
|
|
|
def _since(hours: int) -> datetime:
|
|
return datetime.now(timezone.utc) - timedelta(hours=hours)
|
|
|
|
|
|
def _aware(when: datetime) -> datetime:
|
|
"""A bound as the columns store it. A naive one is read as UTC."""
|
|
return when if when.tzinfo else when.replace(tzinfo=timezone.utc)
|
|
|
|
|
|
@router.get("/summary", response_model=HealthSummary)
|
|
async def read_summary(
|
|
request: Request, controller: FlowControllerDep, session: SessionDep
|
|
) -> Any:
|
|
"""How the engine is doing right now. Always 200, degraded or not."""
|
|
watchdog = getattr(request.app.state, "watchdog", None)
|
|
problems: list[str] = []
|
|
if watchdog is not None and watchdog.degraded:
|
|
problems.append("event loop lagging")
|
|
|
|
queue = await run_in_threadpool(controller.queue_stats)
|
|
if queue.get("oldest_pending_s", 0) > 120:
|
|
problems.append("queue stalled")
|
|
if queue.get("error"):
|
|
problems.append("work queue unreachable")
|
|
|
|
names = await run_in_threadpool(controller.store.list_flows)
|
|
quarantined = controller.quarantined
|
|
paused = set(controller.paused_flows())
|
|
entries = list(controller.loaded.values())
|
|
errored = [e for e in entries if e.status is NodeStatus.ERROR]
|
|
if quarantined:
|
|
problems.append(f"{len(quarantined)} flow(s) quarantined")
|
|
if errored:
|
|
problems.append(f"{len(errored)} node(s) failed to load")
|
|
|
|
statement = (
|
|
select(func.count())
|
|
.select_from(EngineEvent)
|
|
.where(col(EngineEvent.ts) >= _since(24), col(EngineEvent.type) != "audit")
|
|
)
|
|
# The health page polls this every ten seconds; the driver is synchronous.
|
|
failures = await run_in_threadpool(lambda: session.exec(statement).one())
|
|
|
|
return HealthSummary(
|
|
status="degraded" if problems else "ok",
|
|
problems=problems,
|
|
flows={
|
|
"total": len(names),
|
|
"running": len(
|
|
[n for n in names if controller.is_enabled(n) and n not in quarantined]
|
|
),
|
|
"paused": len(paused),
|
|
"quarantined": len(quarantined),
|
|
},
|
|
nodes={"total": len(entries), "error": len(errored)},
|
|
queue=queue,
|
|
loop_lag=(
|
|
watchdog.snapshot()
|
|
if watchdog is not None
|
|
else {"ewma": 0.0, "max_60s": 0.0}
|
|
),
|
|
failures_24h=int(failures),
|
|
)
|
|
|
|
|
|
@router.get("/timeseries", response_model=list[SeriesPoint])
|
|
def read_timeseries(
|
|
session: SessionDep,
|
|
flow: str | None = None,
|
|
node: str | None = None,
|
|
hours: int = 24,
|
|
bucket_s: int = 60,
|
|
) -> Any:
|
|
"""Executions, errors and timings over time, summed across nodes."""
|
|
# ponytail: the fold is in Python — a day is at most 1440 rows per node.
|
|
# date_bin() if the window ever grows past that.
|
|
statement = select(MetricBucket).where(col(MetricBucket.bucket) >= _since(hours))
|
|
if flow:
|
|
statement = statement.where(col(MetricBucket.flow) == flow)
|
|
if node:
|
|
statement = statement.where(col(MetricBucket.node) == node)
|
|
|
|
slices: dict[float, dict[str, float]] = {}
|
|
for row in session.exec(statement.order_by(col(MetricBucket.bucket))):
|
|
stamp = row.bucket.timestamp()
|
|
key = stamp - stamp % max(60, bucket_s)
|
|
point = slices.setdefault(
|
|
key,
|
|
{
|
|
"executions": 0.0,
|
|
"errors": 0.0,
|
|
"messages": 0.0,
|
|
"duration_sum_ms": 0.0,
|
|
"max_ms": 0.0,
|
|
"lag_sum_ms": 0.0,
|
|
"items": 0.0,
|
|
},
|
|
)
|
|
point["executions"] += row.executions
|
|
point["errors"] += row.errors
|
|
point["messages"] += row.messages
|
|
point["duration_sum_ms"] += row.duration_sum_ms
|
|
point["max_ms"] = max(point["max_ms"], row.duration_max_ms)
|
|
point["lag_sum_ms"] += row.lag_sum_ms
|
|
point["items"] += row.items
|
|
|
|
return [
|
|
SeriesPoint(
|
|
ts=ts,
|
|
executions=int(point["executions"]),
|
|
errors=int(point["errors"]),
|
|
messages=int(point["messages"]),
|
|
avg_ms=round(point["duration_sum_ms"] / (point["executions"] or 1), 2),
|
|
max_ms=round(point["max_ms"], 2),
|
|
avg_lag_ms=round(point["lag_sum_ms"] / (point["items"] or 1), 2),
|
|
)
|
|
for ts, point in sorted(slices.items())
|
|
]
|
|
|
|
|
|
@router.get("/flows", response_model=list[FlowRollup])
|
|
def read_flow_rollups(session: SessionDep, hours: int = 24) -> Any:
|
|
"""One row per flow, with a coarse trend of how much it ran."""
|
|
since = _since(hours)
|
|
window = hours * 3600
|
|
start = since.timestamp()
|
|
|
|
rollups: dict[str, dict[str, Any]] = {}
|
|
for row in session.exec(
|
|
select(MetricBucket).where(col(MetricBucket.bucket) >= since)
|
|
):
|
|
entry = rollups.setdefault(
|
|
row.flow,
|
|
{
|
|
"executions": 0,
|
|
"errors": 0,
|
|
"messages": 0,
|
|
"duration_sum_ms": 0.0,
|
|
"lag_sum_ms": 0.0,
|
|
"items": 0,
|
|
"spark": [0] * SPARK_SLICES,
|
|
},
|
|
)
|
|
entry["executions"] += row.executions
|
|
entry["errors"] += row.errors
|
|
entry["messages"] += row.messages
|
|
entry["duration_sum_ms"] += row.duration_sum_ms
|
|
entry["lag_sum_ms"] += row.lag_sum_ms
|
|
entry["items"] += row.items
|
|
slot = min(
|
|
SPARK_SLICES - 1,
|
|
max(0, int((row.bucket.timestamp() - start) / window * SPARK_SLICES)),
|
|
)
|
|
entry["spark"][slot] += row.executions
|
|
|
|
# `.all()` first: a Result has `keys()`, so dict() would read it as a
|
|
# mapping and subscript it.
|
|
last_errors = dict(
|
|
session.exec(
|
|
select(col(EngineEvent.flow), func.max(col(EngineEvent.ts)))
|
|
.where(col(EngineEvent.type) == "node_error", col(EngineEvent.ts) >= since)
|
|
.group_by(col(EngineEvent.flow))
|
|
).all()
|
|
)
|
|
|
|
return [
|
|
FlowRollup(
|
|
flow=flow,
|
|
executions=entry["executions"],
|
|
errors=entry["errors"],
|
|
messages=entry["messages"],
|
|
avg_ms=round(entry["duration_sum_ms"] / (entry["executions"] or 1), 2),
|
|
avg_lag_ms=round(entry["lag_sum_ms"] / (entry["items"] or 1), 2),
|
|
spark=entry["spark"],
|
|
last_error_ts=(
|
|
last_errors[flow].timestamp() if flow in last_errors else None
|
|
),
|
|
)
|
|
for flow, entry in sorted(rollups.items())
|
|
]
|
|
|
|
|
|
@router.get("/runs", response_model=list[RunRow])
|
|
def read_runs(
|
|
session: SessionDep,
|
|
flow: str | None = None,
|
|
status: str | None = None,
|
|
since: datetime | None = None,
|
|
until: datetime | None = None,
|
|
limit: int = 50,
|
|
) -> Any:
|
|
"""Recent cascades, newest first.
|
|
|
|
``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.
|
|
"""
|
|
statement = select(FlowRun).order_by(col(FlowRun.started_at).desc())
|
|
if flow:
|
|
statement = statement.where(col(FlowRun.flow) == flow)
|
|
if status:
|
|
statement = statement.where(col(FlowRun.status) == status)
|
|
if since:
|
|
statement = statement.where(col(FlowRun.started_at) >= _aware(since))
|
|
if until:
|
|
statement = statement.where(col(FlowRun.started_at) < _aware(until))
|
|
return list(session.exec(statement.limit(min(limit, 200))))
|
|
|
|
|
|
@router.get("/events", response_model=list[EventRow])
|
|
def read_events(
|
|
session: SessionDep,
|
|
kind: Literal["failure", "audit"] = "failure",
|
|
flow: str | None = None,
|
|
limit: int = 100,
|
|
) -> Any:
|
|
"""What went wrong, or who changed what. Newest first."""
|
|
statement = select(EngineEvent).order_by(col(EngineEvent.ts).desc())
|
|
if kind == "audit":
|
|
statement = statement.where(col(EngineEvent.type) == "audit")
|
|
else:
|
|
statement = statement.where(col(EngineEvent.type) != "audit")
|
|
if flow:
|
|
statement = statement.where(col(EngineEvent.flow) == flow)
|
|
return list(session.exec(statement.limit(min(limit, 500))))
|
|
|
|
|
|
@router.get("/dead-letter", response_model=list[DeadLetter])
|
|
async def read_dead_letters(controller: FlowControllerDep, limit: int = 50) -> Any:
|
|
"""Work the engine gave up on, which nothing else surfaces."""
|
|
if controller.execution is None:
|
|
return []
|
|
return await run_in_threadpool(controller.execution.queue.dead_letters, limit)
|