app/backend/fluksio/api/routes/observability.py

"""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]
    unhealthy = [e for e in entries if e.health == "down"]
    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)