stroblmeandClaude Opus 5
7b67eb5a8e
A connector that misses one poll turned the health badge amber and the node on the canvas red until it recovered. `LoadedNode.health_since` marks when the status last actually changed — a rotating error detail under an unchanged status does not move it, or a connector retrying with a different errno each poll would never debounce — and `unhealthy_nodes` applies a 60 s floor to it. The floor is on the surfacing, not on the transition: the stored health is the truth the moment a node reports it, so a flow reading node health is never told a stale story. Only going down is held back; recovery clears at once. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CL9zvnnvcp1mvA8o7impxk
509 lines
18 KiB
Python
509 lines
18 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.
|
|
"""
|
|
|
|
import time
|
|
from datetime import UTC, datetime, timedelta
|
|
from typing import Annotated, Any, Literal
|
|
|
|
from fastapi import APIRouter, Depends, Query, Request
|
|
from fastapi.concurrency import run_in_threadpool
|
|
from pydantic import BaseModel, model_validator
|
|
from sqlalchemy import ColumnElement, Integer, cast, func
|
|
from sqlalchemy import select as sa_select
|
|
from sqlmodel import col, select
|
|
|
|
from fluksio import __version__
|
|
from fluksio.api.deps import FlowControllerDep, SessionDep, get_current_user
|
|
from fluksio.api.routes.runs import elapsed_ms
|
|
from fluksio.core.config import settings
|
|
from fluksio.flow.controller import NodeStatus
|
|
from fluksio.flow.pipeline import ADVISORY_ISSUES
|
|
from fluksio.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
|
|
|
|
|
|
def _epoch(column: Any) -> ColumnElement[int]:
|
|
"""Seconds since 1970, as an integer the database can bin on.
|
|
|
|
Binning is integer arithmetic on this rather than a dialect's own date
|
|
function: buckets are whole minutes, so the cast is exact, and the
|
|
expression is the same everywhere.
|
|
"""
|
|
return cast(func.extract("epoch", column), Integer)
|
|
|
|
|
|
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]
|
|
#: What this engine is running. A client ships ahead of the engine it
|
|
#: talks to — a `pip install -U` upgrades one and not the other — and a
|
|
#: route the client knows and the engine does not answers a flat 404. This
|
|
#: is what turns that into a sentence. Absent means older than this field.
|
|
version: str = ""
|
|
|
|
|
|
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
|
|
#: How long it took, or — while it is still going — how long it has been
|
|
#: going: the collector only writes a duration once a cascade finishes.
|
|
duration_ms: float
|
|
deliveries: int
|
|
|
|
@model_validator(mode="after")
|
|
def _running_duration(self) -> "RunRow":
|
|
if self.status == "running" and not self.duration_ms:
|
|
self.duration_ms = elapsed_ms(self.started_at)
|
|
return self
|
|
|
|
|
|
class RunPage(BaseModel):
|
|
data: list[RunRow]
|
|
count: int
|
|
|
|
|
|
class EventRow(BaseModel):
|
|
id: int
|
|
ts: datetime
|
|
type: str
|
|
flow: str
|
|
node: str
|
|
detail: str
|
|
actor: str
|
|
run: str
|
|
|
|
|
|
class DeadLetter(BaseModel):
|
|
id: str
|
|
ts: float
|
|
flow: str
|
|
node: str
|
|
cause: str
|
|
reason: str
|
|
|
|
|
|
def _since(hours: int) -> datetime:
|
|
return datetime.now(UTC) - timedelta(hours=hours)
|
|
|
|
|
|
def _closed_before(stride: int) -> datetime:
|
|
"""The start of the bin `now` falls in — the newest one still filling.
|
|
|
|
A bucket is upserted every flush while its minute runs, so the newest bin
|
|
always holds part of a minute. Drawn, it reads as a fall that never
|
|
happened; excluded, the curve ends on the last bin that is all there.
|
|
"""
|
|
now = int(datetime.now(UTC).timestamp())
|
|
return datetime.fromtimestamp(now // stride * stride, UTC)
|
|
|
|
|
|
#: The most slices one timeseries answers with. `hours=720&bucket_s=60` is
|
|
#: 43 200 points into a single JSON array; a chart draws a few hundred.
|
|
MAX_POINTS = 2000
|
|
|
|
|
|
def _window_hours(hours: int) -> int:
|
|
"""A window the rollups can answer for: an hour at least, retention at most.
|
|
|
|
Zero used to divide by nothing and answer 500, and nothing older than
|
|
retention exists, so a larger window is a scan that can only find less.
|
|
"""
|
|
return max(1, min(hours, settings.OBS_RETENTION_DAYS * 24))
|
|
|
|
|
|
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=UTC)
|
|
|
|
|
|
@router.get("/summary", response_model=HealthSummary)
|
|
async def read_summary(request: Request, controller: FlowControllerDep) -> 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("behind"):
|
|
problems.append(f"engine behind: {queue.get('backlog', 0)} items waiting")
|
|
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]
|
|
# Not `e.health == "down"` directly: a node has to have been down for
|
|
# `HEALTH_FLOOR` before it counts, so a device that misses one poll does
|
|
# not turn this screen amber. Same source the canvas marks nodes from.
|
|
unhealthy = controller.unhealthy_nodes()
|
|
if quarantined:
|
|
problems.append(f"{len(quarantined)} flow(s) quarantined")
|
|
if errored:
|
|
problems.append(f"{len(errored)} node(s) failed to load")
|
|
if unhealthy:
|
|
problems.append(
|
|
f"{len(unhealthy)} node(s) down: "
|
|
f"{', '.join(sorted(e.id for e in unhealthy))}"
|
|
)
|
|
# A node that loads and then fails on every call is not in `errored`, and
|
|
# until this it read as healthy. ponytail: one failure reads degraded for
|
|
# 15 minutes; a counter with decay if that proves noisy.
|
|
now = time.time()
|
|
failing = [
|
|
e
|
|
for e in entries
|
|
if e.last_error_ts is not None and now - e.last_error_ts < 900
|
|
]
|
|
if failing:
|
|
problems.append(
|
|
f"{len(failing)} node(s) failing: "
|
|
f"{', '.join(sorted(e.id for e in failing))}"
|
|
)
|
|
|
|
# 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)}")
|
|
|
|
return HealthSummary(
|
|
status="degraded" if problems else "ok",
|
|
problems=problems,
|
|
flows={
|
|
"total": len(names),
|
|
# What the engine will actually act on. Enabled is not enough:
|
|
# validation stops a flow as surely as quarantine does, and a
|
|
# paused one is holding its messages rather than running them.
|
|
"running": len(
|
|
[
|
|
n
|
|
for n in names
|
|
if controller.is_enabled(n)
|
|
and n not in quarantined
|
|
and n not in paused
|
|
and n not in invalid
|
|
]
|
|
),
|
|
"paused": len(paused),
|
|
"quarantined": len(quarantined),
|
|
"invalid": len(invalid),
|
|
},
|
|
nodes={
|
|
"total": len(entries),
|
|
"error": len(errored),
|
|
"unhealthy": len(unhealthy),
|
|
"failing": len(failing),
|
|
},
|
|
queue=queue,
|
|
loop_lag=(
|
|
watchdog.snapshot()
|
|
if watchdog is not None
|
|
else {"ewma": 0.0, "max_60s": 0.0}
|
|
),
|
|
version=__version__,
|
|
)
|
|
|
|
|
|
@router.get("/timeseries", response_model=list[SeriesPoint])
|
|
def read_timeseries(
|
|
session: SessionDep,
|
|
flow: str | None = None,
|
|
node: str | None = None,
|
|
hours: Annotated[int, Query(ge=0, le=8760)] = 24,
|
|
bucket_s: Annotated[int, Query(ge=1, le=86400)] = 60,
|
|
since: datetime | None = None,
|
|
until: datetime | None = None,
|
|
) -> Any:
|
|
"""Executions, errors and timings over time, summed across nodes.
|
|
|
|
``hours`` measures back from now, which is what a range picker asks for.
|
|
``since`` and ``until`` name a window instead — inclusive and exclusive,
|
|
the same pair ``/runs`` and ``/events`` take — so a chart dragged to a
|
|
span can be re-fetched at that span's own resolution rather than
|
|
magnifying the buckets it already has.
|
|
"""
|
|
hours = _window_hours(hours)
|
|
# The database does the fold: a week of minute rows per node used to cross
|
|
# the wire on every poll, and only the slices need to.
|
|
stride = max(60, bucket_s)
|
|
if since is not None and until is not None:
|
|
# Enough slices to draw with, and never so many that the response is
|
|
# the problem the fold was there to solve.
|
|
span = (_aware(until) - _aware(since)).total_seconds()
|
|
stride = max(stride, int(span // MAX_POINTS) + 1)
|
|
slot = (_epoch(col(MetricBucket.bucket)) // stride * stride).label("slot")
|
|
statement = sa_select(
|
|
slot,
|
|
func.sum(col(MetricBucket.executions)).label("executions"),
|
|
func.sum(col(MetricBucket.errors)).label("errors"),
|
|
func.sum(col(MetricBucket.messages)).label("messages"),
|
|
func.sum(col(MetricBucket.duration_sum_ms)).label("duration_sum_ms"),
|
|
func.max(col(MetricBucket.duration_max_ms)).label("max_ms"),
|
|
func.sum(col(MetricBucket.lag_sum_ms)).label("lag_sum_ms"),
|
|
func.sum(col(MetricBucket.items)).label("items"),
|
|
).where(
|
|
col(MetricBucket.bucket) >= (_aware(since) if since else _since(hours)),
|
|
col(MetricBucket.bucket) < (_aware(until) if until else _closed_before(stride)),
|
|
)
|
|
if flow:
|
|
statement = statement.where(col(MetricBucket.flow) == flow)
|
|
if node:
|
|
statement = statement.where(col(MetricBucket.node) == node)
|
|
|
|
return [
|
|
SeriesPoint(
|
|
ts=float(row.slot),
|
|
executions=int(row.executions),
|
|
errors=int(row.errors),
|
|
messages=int(row.messages),
|
|
avg_ms=round(row.duration_sum_ms / (row.executions or 1), 2),
|
|
max_ms=round(row.max_ms, 2),
|
|
avg_lag_ms=round(row.lag_sum_ms / (row.items or 1), 2),
|
|
)
|
|
for row in session.execute(statement.group_by(slot).order_by(slot))
|
|
]
|
|
|
|
|
|
@router.get("/flows", response_model=list[FlowRollup])
|
|
def read_flow_rollups(
|
|
session: SessionDep, hours: Annotated[int, Query(ge=0, le=8760)] = 24
|
|
) -> Any:
|
|
"""One row per flow, with a coarse trend of how much it ran."""
|
|
hours = _window_hours(hours)
|
|
window = hours * 3600
|
|
# The window ends on the last closed minute rather than on `now`, so every
|
|
# slice it hands back is a whole one. Ending at `now` cut the last slice
|
|
# wherever the request happened to land, and the trend fell off a cliff
|
|
# that was only the clock.
|
|
end = int(datetime.now(UTC).timestamp()) // 60 * 60
|
|
start = float(end - window)
|
|
since = datetime.fromtimestamp(start, UTC)
|
|
|
|
# Binned to the sparkline slice rather than the minute, so a flow costs at
|
|
# most SPARK_SLICES rows however long the window is. The slice is the
|
|
# window over SPARK_SLICES, which for whole hours is whole minutes.
|
|
stride = hours * 60
|
|
origin = int(start)
|
|
slot = (
|
|
(_epoch(col(MetricBucket.bucket)) - origin) // stride * stride + origin
|
|
).label("slot")
|
|
statement = (
|
|
sa_select(
|
|
col(MetricBucket.flow),
|
|
slot,
|
|
func.sum(col(MetricBucket.executions)).label("executions"),
|
|
func.sum(col(MetricBucket.errors)).label("errors"),
|
|
func.sum(col(MetricBucket.messages)).label("messages"),
|
|
func.sum(col(MetricBucket.duration_sum_ms)).label("duration_sum_ms"),
|
|
func.sum(col(MetricBucket.lag_sum_ms)).label("lag_sum_ms"),
|
|
func.sum(col(MetricBucket.items)).label("items"),
|
|
)
|
|
.where(
|
|
col(MetricBucket.bucket) >= since,
|
|
col(MetricBucket.bucket) < datetime.fromtimestamp(float(end), UTC),
|
|
)
|
|
.group_by(col(MetricBucket.flow), slot)
|
|
)
|
|
|
|
rollups: dict[str, dict[str, Any]] = {}
|
|
for row in session.execute(statement):
|
|
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
|
|
# A slot sits a whole number of slices from `since`, so this rounds
|
|
# rather than truncates: a float a hair short would lose a slice. The
|
|
# clamp holds the bucket landing exactly on the far edge in range.
|
|
index = min(
|
|
SPARK_SLICES - 1,
|
|
max(0, round((float(row.slot) - start) / window * SPARK_SLICES)),
|
|
)
|
|
entry["spark"][index] += 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=RunPage)
|
|
def read_runs(
|
|
session: SessionDep,
|
|
flow: str | None = None,
|
|
status: str | None = None,
|
|
since: datetime | None = None,
|
|
until: datetime | None = None,
|
|
limit: Annotated[int, Query(ge=1, le=200)] = 50,
|
|
) -> Any:
|
|
"""Recent cascades, newest first, and how many there were in total.
|
|
|
|
``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.
|
|
"""
|
|
filters: list[ColumnElement[bool]] = []
|
|
if flow:
|
|
filters.append(col(FlowRun.flow) == flow)
|
|
if status:
|
|
filters.append(col(FlowRun.status) == status)
|
|
if since:
|
|
filters.append(col(FlowRun.started_at) >= _aware(since))
|
|
if until:
|
|
filters.append(col(FlowRun.started_at) < _aware(until))
|
|
|
|
capped = min(limit, 200)
|
|
statement = select(FlowRun).where(*filters).order_by(col(FlowRun.started_at).desc())
|
|
rows = list(session.exec(statement.limit(capped)))
|
|
# A short page is its own total. The lists poll their whole range every
|
|
# thirty seconds, and counting on each of those would hand back what
|
|
# binning the metrics just saved — for a number that only ever says
|
|
# "there is more here than fits".
|
|
count = (
|
|
len(rows)
|
|
if len(rows) < capped
|
|
else int(
|
|
session.exec(
|
|
select(func.count()).select_from(FlowRun).where(*filters)
|
|
).one()
|
|
)
|
|
)
|
|
return {"data": rows, "count": count}
|
|
|
|
|
|
@router.get("/events", response_model=list[EventRow])
|
|
def read_events(
|
|
session: SessionDep,
|
|
kind: Literal["failure", "audit"] = "failure",
|
|
flow: str | None = None,
|
|
run: str | None = None,
|
|
since: datetime | None = None,
|
|
until: datetime | None = None,
|
|
limit: Annotated[int, Query(ge=1, le=500)] = 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.
|
|
``run`` narrows to one run — a batch run's id, or the journaled item a live
|
|
cascade came from. Rows recorded before the column existed carry none, so
|
|
an old failure answers no run at all rather than the wrong one.
|
|
"""
|
|
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 run:
|
|
statement = statement.where(col(EngineEvent.run) == run)
|
|
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: Annotated[int, Query(ge=1, le=200)] = 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)
|