stroblme
857a5a57d8
A dependency loop is flagged on the canvas and was invisible everywhere else: /observability/summary answered "ok" with an empty problems list while the published flow could not run at all. It now reports the flows validation blocks, and the brain graph carries the reason on each neuron the issue names so the view built to find broken wiring can show it. Node errors stay counted once, as the nodes that failed to load, and an advisory like an unauthenticated webhook marks nothing — it is worth saying, but the flow still runs.
355 lines
11 KiB
Python
355 lines
11 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 ADVISORY_ISSUES, 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")
|
|
|
|
# What the canvas flags on a flow — a dependency loop, an input nothing
|
|
# feeds — stops that flow running just as surely as a node that will not
|
|
# load, and until now this screen was the one place it did not show.
|
|
# `node_error` is left out: those are the nodes already counted above.
|
|
invalid = sorted(
|
|
{
|
|
issue.flow
|
|
for issue in controller.issues
|
|
if issue.flow
|
|
and issue.code != "node_error"
|
|
and issue.code not in ADVISORY_ISSUES
|
|
}
|
|
)
|
|
if invalid:
|
|
problems.append(f"{len(invalid)} flow(s) cannot run: {', '.join(invalid)}")
|
|
|
|
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),
|
|
"invalid": len(invalid),
|
|
},
|
|
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,
|
|
since: datetime | None = None,
|
|
until: datetime | None = None,
|
|
limit: int = 100,
|
|
) -> Any:
|
|
"""What went wrong, or who changed what. Newest first.
|
|
|
|
``since`` is inclusive and ``until`` exclusive, the same window ``/runs``
|
|
takes, so a list can cover the span the charts beside it are drawn from.
|
|
"""
|
|
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)
|
|
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))))
|
|
|
|
|
|
@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)
|