From af3ba51571de85721dc58cade901b06cb720693b Mon Sep 17 00:00:00 2001 From: stroblme Date: Sun, 16 Aug 2026 22:29:32 +0200 Subject: [PATCH] Keep the engine's own history, and a screen that reads it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A second bus subscriber folds executions, errors, timings and queue lag into per-minute rollups, keeps failures with their traceback and an audit trail of who published what, and records one row per cascade — manual runs and previews included, under an id of their own that writes no idempotency markers. Read back through /observability/*, which always answers 200 so a degraded engine still renders its own health screen. Also fixes two things found on the way: node-health alerts read `status` where the engine publishes `health`, so a device dropping never alerted anyone, and the Redis queue reported `parked: 0` whatever was held. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_017MeiWk3Yq12n2pTvnQWYvt --- NOTEPAD.md | 10 +- ROADMAP.md | 10 + ...4_add_observability_rollups_events_and_.py | 77 +++ backend/app/api/main.py | 2 + backend/app/api/routes/flows.py | 36 +- backend/app/api/routes/modules.py | 23 +- backend/app/api/routes/observability.py | 310 ++++++++++++ backend/app/core/config.py | 2 + backend/app/flow/alerts.py | 4 +- backend/app/flow/executor.py | 38 ++ backend/app/flow/metrics.py | 356 ++++++++++++++ backend/app/flow/pipeline.py | 54 ++- backend/app/flow/queue.py | 37 +- backend/app/main.py | 5 + backend/app/mcp/server.py | 42 ++ backend/app/models.py | 81 ++++ .../tests/api/routes/test_observability.py | 104 ++++ backend/tests/flow/test_alerts.py | 12 +- backend/tests/test_metrics.py | 149 ++++++ frontend/src/client/schemas.gen.ts | 261 ++++++++++ frontend/src/client/sdk.gen.ts | 136 +++++- frontend/src/client/types.gen.ts | 110 +++++ frontend/src/components/Common/UplotChart.tsx | 163 +++++++ .../src/components/Dashboard/ChartWidget.tsx | Bin 7726 -> 3788 bytes .../src/components/Flow/MessageSparkline.tsx | 2 +- frontend/src/components/Flow/liveStore.ts | 49 ++ frontend/src/components/Flow/useFlowSocket.ts | 79 ++- .../src/components/Sidebar/AppSidebar.tsx | 2 + frontend/src/routeTree.gen.ts | 21 + frontend/src/routes/_layout/health.tsx | 457 ++++++++++++++++++ 30 files changed, 2610 insertions(+), 22 deletions(-) create mode 100644 backend/app/alembic/versions/087c44e16304_add_observability_rollups_events_and_.py create mode 100644 backend/app/api/routes/observability.py create mode 100644 backend/app/flow/metrics.py create mode 100644 backend/tests/api/routes/test_observability.py create mode 100644 backend/tests/test_metrics.py create mode 100644 frontend/src/components/Common/UplotChart.tsx create mode 100644 frontend/src/routes/_layout/health.tsx diff --git a/NOTEPAD.md b/NOTEPAD.md index 2127ca5..27d156d 100644 --- a/NOTEPAD.md +++ b/NOTEPAD.md @@ -29,7 +29,6 @@ is what M4 still waits on, together with porting the flows. ### Bugs found while building the screens - BUG/API: `POST /alerts/test/{channel}` always answers 200. `AlertManager.send()` catches and logs every delivery failure, so the alerts screen's Test button cannot tell a working channel from a broken one — the one thing it exists for. Let `send()` raise or return a result on the test path. -- BUG/API: `RedisWorkQueue.stats()` hard-codes `"parked": 0`, so held or stranded work never shows on the health endpoint. This is why a rebuild stranding parked items went unnoticed until it was looked for. - BUG/FLOW: deleting a flow leaves its `pipeline:{flow}.*` Redis keys behind, and renaming one does not migrate them — the live instance carries `pipeline:__history__:dashboar.test` beside the correct `dashboard.test`. One cleanup on the delete/rename path covers both. - CHORE/API: revoking an OAuth client does not invalidate access tokens already issued; they are stateless JWTs valid up to `MCP_TOKEN_EXPIRE_MINUTES`. Immediate revocation means `app/mcp/http.py` checking the client row still exists. - CHORE/FLOW: `Pipeline.trigger`'s docstring says a paused flow still publishes so the value shows on the canvas. True only without a queue; with one the item parks before `apply_outputs` and nothing shows. Docstring and behaviour disagree. @@ -44,6 +43,15 @@ is what M4 still waits on, together with porting the flows. - FEAT/API: `POST /modules/apply` rebuilds the whole pipeline so a node that could not import its package stops being red. That resubscribes every MQTT node in the deployment; a targeted rebuild of the flows that actually failed to load would be gentler. - CHORE/FLOW: a node's return value now round-trips through JSON, so tuples arrive downstream as lists and anything non-JSON is an explicit error. That is the message contract, but flows written before this may notice. +### Engine history + +- CHORE/FLOW: a rate-limit flush gets no run record — it is the tail of the run that scheduled it, and there is no id linking the two. A flush that fails therefore shows as a failure with no run beside it. +- CHORE/FLOW: `Pipeline.flush` releasing a held value runs its cascade without a run id, so those executions land in the minute rollups but in no run. Threading the scheduling run's id through the queue item would close it. +- CHORE/API: the metrics collector is a bus subscriber, so a storm that overflows the bus queue undercounts. The events dropped are the same ones the websocket drops; exact accounting would need the collector to be fed from the engine rather than the bus. +- CHORE/API: `/observability/summary` reports the work queue's `depth` as the Redis stream length, which is the journal size (capped at `STREAM_MAXLEN`) rather than a backlog. The health screen shows `pending` instead; the field name still invites the wrong reading. +- FEAT/UI: the health screen's window is fixed at 24 hours and the charts fold minute buckets in Python. A range picker (and `date_bin()` behind it) is the next step if anyone wants a week. +- CHORE/FLOW: run records for a deleted flow stay until the retention window passes, so a flow that no longer exists keeps appearing in the history. Deliberate — it is a record of what ran — but `forget_flow` could offer to clear it. + ### Dashboard follow-ups - BUG/UI: ensure dashboard wallpanel (read-only) links hot reload automatically on dashboard changes diff --git a/ROADMAP.md b/ROADMAP.md index 6305564..f663982 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -107,6 +107,12 @@ external interfaces. See `docs/architecture/structure.canvas` → *Backend – M a dead consumer never acknowledged; nodes that reach outside are skipped on a redelivery they already ran. Long-lived worker pools replace the per-wave executors, and a delay now waits in the queue rather than on a worker thread +- [x] Engine history in Postgres: a second bus subscriber folds executions, errors, + timings and queue lag into per-minute rollups, keeps failures with their + traceback and an audit trail of who published what, and records one row per + cascade — including the manual runs and previews that never went through the + queue. Read back through `/observability/*`, which always answers 200 so a + degraded engine still renders, and pruned on a retention window - [ ] Test nodes: a small node dragged onto an existing one, smoke or unit, blocking deployment on failure - [ ] User management scoped per flow and per data set @@ -169,6 +175,10 @@ React + Vite, primarily desktop but usable on mobile. See `docs/architecture/str channels/rules each get a sidebar page, and the OAuth clients an agent registers are listed and revocable under Admin — which needed its management endpoints written first +- [x] Health screen: how the engine is doing now (nodes, flows, queue, loop lag) + over what it has been doing all day — throughput and failure charts, a + per-flow table, the recent cascades, failures that expand to their + traceback, dead-lettered work and the audit trail - [x] Mobile-friendly canvas: touch connect, full-screen node panel - [ ] Installable as a PWA (`vite-plugin-pwa`) diff --git a/backend/app/alembic/versions/087c44e16304_add_observability_rollups_events_and_.py b/backend/app/alembic/versions/087c44e16304_add_observability_rollups_events_and_.py new file mode 100644 index 0000000..a35f29a --- /dev/null +++ b/backend/app/alembic/versions/087c44e16304_add_observability_rollups_events_and_.py @@ -0,0 +1,77 @@ +"""Add observability rollups, events and runs + +Revision ID: 087c44e16304 +Revises: 59e2606ce144 +Create Date: 2026-08-16 19:55:10.772691 + +""" +from alembic import op +import sqlalchemy as sa +import sqlmodel.sql.sqltypes + + +# revision identifiers, used by Alembic. +revision = '087c44e16304' +down_revision = '59e2606ce144' +branch_labels = None +depends_on = None + + +def upgrade(): + # ### commands auto generated by Alembic - please adjust! ### + op.create_table('engine_event', + sa.Column('id', sa.Integer(), nullable=False), + sa.Column('ts', sa.DateTime(timezone=True), nullable=False), + sa.Column('type', sqlmodel.sql.sqltypes.AutoString(length=32), nullable=False), + sa.Column('flow', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('node', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('detail', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('actor', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.PrimaryKeyConstraint('id') + ) + op.create_index(op.f('ix_engine_event_ts'), 'engine_event', ['ts'], unique=False) + op.create_index(op.f('ix_engine_event_type'), 'engine_event', ['type'], unique=False) + op.create_table('flow_run', + sa.Column('id', sqlmodel.sql.sqltypes.AutoString(length=64), nullable=False), + sa.Column('flow', sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False), + sa.Column('source', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('started_at', sa.DateTime(timezone=True), nullable=False), + sa.Column('finished_at', sa.DateTime(timezone=True), nullable=True), + sa.Column('status', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('nodes', sa.Integer(), nullable=False), + sa.Column('errors', sa.Integer(), nullable=False), + sa.Column('duration_ms', sa.Float(), nullable=False), + sa.Column('deliveries', sa.Integer(), nullable=False), + sa.PrimaryKeyConstraint('id') + ) + op.create_index(op.f('ix_flow_run_flow'), 'flow_run', ['flow'], unique=False) + op.create_index(op.f('ix_flow_run_started_at'), 'flow_run', ['started_at'], unique=False) + op.create_table('metric_minute', + sa.Column('flow', sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False), + sa.Column('node', sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False), + sa.Column('bucket', sa.DateTime(timezone=True), nullable=False), + sa.Column('executions', sa.Integer(), nullable=False), + sa.Column('errors', sa.Integer(), nullable=False), + sa.Column('messages', sa.Integer(), nullable=False), + sa.Column('duration_sum_ms', sa.Float(), nullable=False), + sa.Column('duration_max_ms', sa.Float(), nullable=False), + sa.Column('lag_sum_ms', sa.Float(), nullable=False), + sa.Column('lag_max_ms', sa.Float(), nullable=False), + sa.Column('items', sa.Integer(), nullable=False), + sa.PrimaryKeyConstraint('flow', 'node', 'bucket') + ) + op.create_index(op.f('ix_metric_minute_bucket'), 'metric_minute', ['bucket'], unique=False) + # ### end Alembic commands ### + + +def downgrade(): + # ### commands auto generated by Alembic - please adjust! ### + op.drop_index(op.f('ix_metric_minute_bucket'), table_name='metric_minute') + op.drop_table('metric_minute') + op.drop_index(op.f('ix_flow_run_started_at'), table_name='flow_run') + op.drop_index(op.f('ix_flow_run_flow'), table_name='flow_run') + op.drop_table('flow_run') + op.drop_index(op.f('ix_engine_event_type'), table_name='engine_event') + op.drop_index(op.f('ix_engine_event_ts'), table_name='engine_event') + op.drop_table('engine_event') + # ### end Alembic commands ### diff --git a/backend/app/api/main.py b/backend/app/api/main.py index ad9067e..af9d48d 100644 --- a/backend/app/api/main.py +++ b/backend/app/api/main.py @@ -8,6 +8,7 @@ from app.api.routes import ( messages, modules, oauth, + observability, private, secrets, users, @@ -25,6 +26,7 @@ api_router.include_router(alerts.router) api_router.include_router(dashboards.router) api_router.include_router(messages.router) api_router.include_router(modules.router) +api_router.include_router(observability.router) # Always mounted so the generated SDK stays the same shape; the endpoints # themselves refuse to work unless MCP is switched on. api_router.include_router(oauth.router) diff --git a/backend/app/api/routes/flows.py b/backend/app/api/routes/flows.py index 6994156..b6e725e 100644 --- a/backend/app/api/routes/flows.py +++ b/backend/app/api/routes/flows.py @@ -1,6 +1,7 @@ """The flow API. Everything the editor can do is available here first.""" import asyncio +import time from typing import Any from fastapi import ( @@ -14,7 +15,12 @@ from fastapi.concurrency import run_in_threadpool from pydantic import BaseModel from sqlmodel import Session -from app.api.deps import FlowControllerDep, get_current_user, user_from_token +from app.api.deps import ( + CurrentUser, + FlowControllerDep, + get_current_user, + user_from_token, +) from app.core.db import engine from app.flow.controller import FlowController from app.flow.dashboards import DashboardStore @@ -184,6 +190,19 @@ def _read_flow(controller: FlowController, name: str) -> FlowDef: raise HTTPException(status_code=404, detail=f"No flow named '{name}'") +def _audit(action: str, flow: str, user: CurrentUser) -> None: + """Record who changed what. The collector writes it down; the bus carries it.""" + event_bus.publish( + { + "type": "audit", + "action": action, + "flow": flow, + "user": user.email, + "ts": time.time(), + } + ) + + def _source_ref(definition: FlowDef, node_id: str) -> str | None: """The library source this node runs, if it is a shared one.""" node = next((n for n in definition.nodes if n.id == node_id), None) @@ -323,6 +342,7 @@ async def publish_flow( name: str, body: PublishRequest, controller: FlowControllerDep, + user: CurrentUser, ) -> Any: """Deploy the unpublished changes: the engine picks them up from here.""" _read_flow(controller, name) @@ -339,6 +359,7 @@ async def publish_flow( status_code=409, detail={"message": str(exc), "current_version": exc.current}, ) + _audit("published", name, user) await controller.reload() return _detail(controller, published) @@ -364,12 +385,15 @@ async def discard_draft(name: str, controller: FlowControllerDep) -> Any: @router.delete("/{name}", response_model=Message) -async def delete_flow(name: str, controller: FlowControllerDep) -> Any: +async def delete_flow( + name: str, controller: FlowControllerDep, user: CurrentUser +) -> Any: """Delete a flow and everything in it.""" try: await run_in_threadpool(controller.store.delete_flow, name) except FlowNotFound: raise HTTPException(status_code=404, detail=f"No flow named '{name}'") + _audit("deleted", name, user) # Its files are gone; its values and queued work would otherwise linger. await run_in_threadpool(controller.forget_flow, name) await controller.reload() @@ -510,18 +534,22 @@ async def unshare_node( @router.post("/{name}/start", response_model=FlowDetail) -async def start_flow(name: str, controller: FlowControllerDep) -> Any: +async def start_flow( + name: str, controller: FlowControllerDep, user: CurrentUser +) -> Any: """Let the engine run this flow again.""" _read_flow(controller, name) await controller.set_enabled(name, True) + _audit("started", name, user) return _detail(controller, _read_flow(controller, name)) @router.post("/{name}/stop", response_model=FlowDetail) -async def stop_flow(name: str, controller: FlowControllerDep) -> Any: +async def stop_flow(name: str, controller: FlowControllerDep, user: CurrentUser) -> Any: """Take this flow off the engine: no subscriptions, schedules or webhooks.""" _read_flow(controller, name) await controller.set_enabled(name, False) + _audit("stopped", name, user) return _detail(controller, _read_flow(controller, name)) diff --git a/backend/app/api/routes/modules.py b/backend/app/api/routes/modules.py index 4ad03ec..43aa363 100644 --- a/backend/app/api/routes/modules.py +++ b/backend/app/api/routes/modules.py @@ -6,13 +6,20 @@ the worker pool retires its processes, and the next node call picks up the new packages. """ +import time from typing import Any from fastapi import APIRouter, Depends, HTTPException from fastapi.concurrency import run_in_threadpool -from app.api.deps import FlowControllerDep, WorkerPoolDep, get_current_user +from app.api.deps import ( + CurrentUser, + FlowControllerDep, + WorkerPoolDep, + get_current_user, +) from app.flow import modules +from app.flow.events import event_bus from app.flow.schemas import ApplyRequest, ApplyResult, ModulesInfo router = APIRouter( @@ -28,7 +35,10 @@ async def read_modules(controller: FlowControllerDep) -> Any: @router.post("/apply", response_model=ApplyResult) async def apply_modules( - body: ApplyRequest, controller: FlowControllerDep, pool: WorkerPoolDep + body: ApplyRequest, + controller: FlowControllerDep, + pool: WorkerPoolDep, + user: CurrentUser, ) -> Any: """Install exactly these requirements, then hand them to the workers. @@ -42,6 +52,15 @@ async def apply_modules( detail=output or "These requirements could not be installed", ) await run_in_threadpool(controller.store.write_requirements, body.requirements) + event_bus.publish( + { + "type": "audit", + "action": "installed modules", + "flow": "", + "user": user.email, + "ts": time.time(), + } + ) # Retire the workers first, so the rebuild compiles every node against the # packages that were just installed — a node that could not import one is # the reason this was called, and it stays red until it is built again. diff --git a/backend/app/api/routes/observability.py b/backend/app/api/routes/observability.py new file mode 100644 index 0000000..7222bc3 --- /dev/null +++ b/backend/app/api/routes/observability.py @@ -0,0 +1,310 @@ +"""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) + + +@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") + + failures = session.exec( + select(func.count()) + .select_from(EngineEvent) + .where(col(EngineEvent.ts) >= _since(24), col(EngineEvent.type) != "audit") + ).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, + limit: int = 50, +) -> Any: + """Recent cascades, newest first.""" + 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) + 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) diff --git a/backend/app/core/config.py b/backend/app/core/config.py index d221a2d..e653012 100644 --- a/backend/app/core/config.py +++ b/backend/app/core/config.py @@ -63,6 +63,8 @@ class Settings(BaseSettings): # node sets its own. Long enough for a slow HTTP call, short enough that a # runaway loop is not a wedged flow. FLOW_NODE_TIMEOUT: float = 30.0 + # How long the engine's own metrics, events and run records are kept. + OBS_RETENTION_DAYS: int = 30 # Without a Redis host the engine keeps its state in memory. REDIS_HOST: str | None = None REDIS_PORT: int = 6379 diff --git a/backend/app/flow/alerts.py b/backend/app/flow/alerts.py index df9f894..11c42bc 100644 --- a/backend/app/flow/alerts.py +++ b/backend/app/flow/alerts.py @@ -102,7 +102,7 @@ def describe(event: dict[str, Any]) -> Alert | None: node=node, ) if kind == "node_health": - if event.get("status") != "down": + if event.get("health") != "down": return None return Alert( title=f"{node or 'A node'} lost its connection", @@ -253,7 +253,7 @@ class AlertManager: if muted_until is not None: if now < muted_until: # Still flapping — push the window out and stay quiet. - if event.get("status") == "down": + if event.get("health") == "down": self._flapping[key] = now + FLAP_WINDOW_S return True del self._flapping[key] diff --git a/backend/app/flow/executor.py b/backend/app/flow/executor.py index ece465c..1a14ca9 100644 --- a/backend/app/flow/executor.py +++ b/backend/app/flow/executor.py @@ -260,6 +260,8 @@ class ExecutionService: if item.kind == "flush": # A rate-limit window ended; nothing to replay, only to let out. + # ponytail: no run record for a flush — it is the tail of the run + # that scheduled it, not a run of its own. pipeline.flush(node) return True @@ -268,10 +270,46 @@ class ExecutionService: logger.debug("Guard no longer holds for '%s', dropped", item.node) return True + # Only here is the item certain to run, which is what a run record is. + now = time.time() + self._publish( + { + "type": "cascade_started", + "run": item.entry_id, + "flow": item.flow, + "node": item.node, + "cause": item.cause, + "deliveries": item.deliveries, + "ts": now, + } + ) + if item.deliveries == 1: + # A redelivery waited for the reaper, not for the engine. + self._publish( + { + "type": "work_latency", + "flow": item.flow, + "node": item.node, + "lag_ms": max( + 0.0, + (now - max(item.enqueued_at, item.not_before)) * 1000, + ), + "ts": now, + } + ) + pipeline.apply_outputs(node, item.outputs or None) pipeline.run_downstream( node, entry_id=item.entry_id, replay=item.deliveries > 1 ) + self._publish( + { + "type": "cascade_finished", + "run": item.entry_id, + "flow": item.flow, + "ts": time.time(), + } + ) return True # ------------------------------------------------------------------------- diff --git a/backend/app/flow/metrics.py b/backend/app/flow/metrics.py new file mode 100644 index 0000000..8d438dd --- /dev/null +++ b/backend/app/flow/metrics.py @@ -0,0 +1,356 @@ +"""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 datetime, timedelta, timezone +from typing import Any + +from sqlalchemy import delete, func, update +from sqlalchemy.dialects.postgresql import insert +from sqlmodel import Session, col + +from app.core.config import settings +from app.core.db import engine +from app.flow.events import EventBus +from app.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 +#: 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) -> datetime: + return datetime.fromtimestamp(ts, timezone.utc).replace(second=0, microsecond=0) + + +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, datetime], 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. + self._tracebacks: dict[tuple[str, str], 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 asyncio.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 ""), + str(event.get("node") or ""), + _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] = str(event.get("text") or "") + 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, "") + error = str(event.get("error") or "") + self._pending.append( + EngineEvent( + ts=datetime.fromtimestamp(ts, timezone.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, timezone.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, timezone.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, timezone.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, timezone.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 ""), + "source": str(event.get("cause") or ""), + "started_ts": ts, + "started_at": datetime.fromtimestamp(ts, timezone.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") + 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. + + def _write( + self, + buckets: dict[tuple[str, str, datetime], 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=minute, **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 + } + | { + name: func.greatest( + 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(timezone.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) + ) diff --git a/backend/app/flow/pipeline.py b/backend/app/flow/pipeline.py index 6465b02..cb50c3e 100644 --- a/backend/app/flow/pipeline.py +++ b/backend/app/flow/pipeline.py @@ -11,6 +11,7 @@ from __future__ import annotations import logging import threading import time +import uuid from collections import deque from collections.abc import Iterator from concurrent.futures import Future, ThreadPoolExecutor, wait @@ -28,6 +29,10 @@ from app.flow.state import MemoryState, StateBackend logger = logging.getLogger(__name__) +#: A run that never went through the queue. It still gets a run record, so a +#: manual run shows up in the history — but it is no one's idempotency key. +MANUAL_RUN_PREFIX = "manual-" + class ValidationIssue(BaseModel): """A problem that keeps a flow from running correctly.""" @@ -589,9 +594,15 @@ class Pipeline: result = node.execute(inputs) self.publish_log(node, collected, "") - if entry_id and not node.idempotent and self._queue is not None: + if ( + entry_id + and not entry_id.startswith(MANUAL_RUN_PREFIX) + and not node.idempotent + and self._queue is not None + ): # Written after the fact: a crash between the side effect and - # this marker is the one window at-least-once cannot close. + # this marker is the one window at-least-once cannot close. A + # manual run has nothing to be redelivered, so it writes none. try: self._queue.mark_done(entry_id, node.id) except Exception as exc: @@ -632,6 +643,7 @@ class Pipeline: # which is a different thing to show than one that emitted. "outputs": len(result or {}), "duration_ms": round((time.perf_counter() - started) * 1000, 2), + "run": entry_id, "ts": time.time(), } ) @@ -648,6 +660,7 @@ class Pipeline: "flow": node.flow, "node": node.id, "error": f"{type(exc).__name__}: {exc}", + "run": entry_id, "ts": time.time(), } ) @@ -896,8 +909,40 @@ class Pipeline: self._enqueue_cascade(node, outputs) return state + return self._run_here(node, outputs) + + def _run_here( + self, node: Node, outputs: dict[str, Any] | None, cause: str = "manual" + ) -> StateBackend: + """Run a cascade in this thread, under a run id of its own. + + The queued path gets its run id from the journal entry. A run that never + went through the queue still belongs in the history, so it makes one — + marked as such, because it is no one's idempotency key. + """ + run_id = f"{MANUAL_RUN_PREFIX}{uuid.uuid4().hex[:12]}" + self._publish( + { + "type": "cascade_started", + "run": run_id, + "flow": node.flow, + "node": node.id, + "cause": cause, + "deliveries": 1, + "ts": time.time(), + } + ) self.apply_outputs(node, outputs) - return self.run_downstream(node) + state = self.run_downstream(node, entry_id=run_id) + self._publish( + { + "type": "cascade_finished", + "run": run_id, + "flow": node.flow, + "ts": time.time(), + } + ) + return state def publish( self, values: dict[str, Any], source: ValueSource | None = None @@ -1008,8 +1053,7 @@ class Pipeline: } ) # Losing the value outright would be worse than running it here. - self.apply_outputs(node, outputs) - self.run_downstream(node) + self._run_here(node, outputs, cause="external") def values(self, flow: str | None = None) -> dict[str, dict[str, Any]]: """Last value and timestamp of every message, optionally one flow's.""" diff --git a/backend/app/flow/queue.py b/backend/app/flow/queue.py index c245864..cb1f781 100644 --- a/backend/app/flow/queue.py +++ b/backend/app/flow/queue.py @@ -53,6 +53,8 @@ class WorkItem: :param entry_id: Set by the queue on claim; stable across redeliveries, which is what makes it usable as an idempotency key. :param deliveries: How many times this item has been handed out. + :param enqueued_at: When the item was made. Carried rather than read off + the entry id, because only the Redis queue has timestamps in its ids. """ kind: str @@ -65,6 +67,7 @@ class WorkItem: guard_value: str = "" entry_id: str = "" deliveries: int = 1 + enqueued_at: float = field(default_factory=time.time) def to_fields(self) -> dict[str, str]: return { @@ -76,6 +79,7 @@ class WorkItem: "not_before": str(self.not_before), "guard_key": self.guard_key, "guard_value": self.guard_value, + "enqueued_at": str(self.enqueued_at), } @classmethod @@ -93,6 +97,7 @@ class WorkItem: guard_value=fields.get("guard_value", ""), entry_id=entry_id, deliveries=deliveries, + enqueued_at=float(fields.get("enqueued_at") or time.time()), ) @@ -147,6 +152,10 @@ class WorkQueue(ABC): def stats(self) -> dict[str, Any]: """Queue depth and age, for the health endpoint.""" + @abstractmethod + def dead_letters(self, count: int = 50) -> list[dict[str, Any]]: + """What was given up on, newest first.""" + @abstractmethod def mark_done(self, entry_id: str, node: str) -> None: """Record that a side effect already happened for this delivery.""" @@ -251,6 +260,10 @@ class MemoryWorkQueue(WorkQueue): "durable": False, } + def dead_letters(self, count: int = 50) -> list[dict[str, Any]]: + """Nothing is kept: a dropped item here died with the process.""" + return [] + def mark_done(self, entry_id: str, node: str) -> None: with self._lock: self._done.add((entry_id, node)) @@ -416,15 +429,37 @@ class RedisWorkQueue(WorkQueue): ) if records: oldest = records[0]["time_since_delivered"] / 1000.0 + parked = sum( + cast(int, self._redis.llen(key)) + for key in self._redis.scan_iter(f"{self._ns}:__parked__:*") + ) return { "depth": cast(int, self._redis.xlen(self._stream)), "pending": count, "delayed": cast(int, self._redis.zcard(self._delayed_key)), - "parked": 0, + "parked": parked, "oldest_pending_s": round(oldest, 1), "durable": True, } + def dead_letters(self, count: int = 50) -> list[dict[str, Any]]: + """What was set aside, newest first. The entry id carries the time.""" + entries = cast( + list[tuple[str, dict[str, str]]], + self._redis.xrevrange(self._dead_key, count=count), + ) + return [ + { + "id": entry_id, + "ts": int(entry_id.split("-")[0]) / 1000.0, + "flow": fields.get("flow", ""), + "node": fields.get("node", ""), + "cause": fields.get("cause", ""), + "reason": fields.get("reason", ""), + } + for entry_id, fields in entries + ] + def mark_done(self, entry_id: str, node: str) -> None: # An hour outlives any redelivery; after that the marker is noise. self._redis.set(self._done_key(entry_id, node), "1", ex=3600) diff --git a/backend/app/main.py b/backend/app/main.py index 18b9fc7..7c1abd0 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -20,6 +20,7 @@ from app.flow.controller import FlowController from app.flow.dashboards import DashboardStore from app.flow.events import event_bus from app.flow.executor import ExecutionService +from app.flow.metrics import MetricsCollector from app.flow.nodes.http import close_shared_client from app.flow.plugins import load_plugins from app.flow.queue import MemoryWorkQueue, RedisWorkQueue, WorkQueue @@ -107,6 +108,9 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: app.state.watchdog = watchdog watchdog_task = asyncio.create_task(watchdog.run(), name="loop-watchdog") alerts_task = asyncio.create_task(alerts.run(), name="alert-manager") + metrics_task = asyncio.create_task( + MetricsCollector(event_bus).run(), name="metrics-collector" + ) await controller.start() try: # A mounted sub-app gets no lifespan of its own, so the MCP session @@ -116,6 +120,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: finally: watchdog_task.cancel() alerts_task.cancel() + metrics_task.cancel() await controller.stop() pool.stop() close_shared_client() diff --git a/backend/app/mcp/server.py b/backend/app/mcp/server.py index fea994d..59bd56d 100644 --- a/backend/app/mcp/server.py +++ b/backend/app/mcp/server.py @@ -265,3 +265,45 @@ async def apply_modules(requirements: str) -> Any: manifest that does not resolve changes nothing. """ return await _call("POST", "/modules/apply", json={"requirements": requirements}) + + +# ----------------------------------------------------------------------------- +# Observability +# ----------------------------------------------------------------------------- + + +@mcp.tool() +async def get_health() -> Any: + """How the engine is doing: flows, nodes, queue, loop lag and recent failures.""" + return await _call("GET", "/observability/summary") + + +@mcp.tool() +async def get_metrics( + flow: str | None = None, node: str | None = None, hours: int = 24 +) -> Any: + """Executions, errors and timings per minute over the last few hours.""" + params: dict[str, Any] = {"hours": hours} + if flow: + params["flow"] = flow + if node: + params["node"] = node + return await _call("GET", "/observability/timeseries", params=params) + + +@mcp.tool() +async def list_failures(flow: str | None = None, limit: int = 50) -> Any: + """Recent failures with their tracebacks, newest first.""" + params: dict[str, Any] = {"kind": "failure", "limit": limit} + if flow: + params["flow"] = flow + return await _call("GET", "/observability/events", params=params) + + +@mcp.tool() +async def list_runs(flow: str | None = None, limit: int = 50) -> Any: + """Recent cascades: what triggered them, how long they took, how they ended.""" + params: dict[str, Any] = {"limit": limit} + if flow: + params["flow"] = flow + return await _call("GET", "/observability/runs", params=params) diff --git a/backend/app/models.py b/backend/app/models.py index ee1cb6f..7653873 100644 --- a/backend/app/models.py +++ b/backend/app/models.py @@ -198,3 +198,84 @@ class OAuthAuthorizeRequest(SQLModel): class OAuthAuthorizeResponse(SQLModel): redirect_url: str + + +# ----------------------------------------------------------------------------- +# Observability +# +# The engine's own history: what ran, how long it took, and what went wrong. +# Rolled up per minute rather than kept per execution — a node firing every +# second is 86 400 rows a day raw, and nobody reads a row of that. +# ----------------------------------------------------------------------------- + + +class MetricBucket(SQLModel, table=True): + """One node's minute: how much ran, how long it took, how late it was.""" + + __tablename__ = "metric_minute" + + flow: str = Field(primary_key=True, max_length=255) + node: str = Field(primary_key=True, max_length=255) + bucket: datetime = Field( + primary_key=True, + index=True, + sa_type=DateTime(timezone=True), # type: ignore + ) + executions: int = 0 + errors: int = 0 + #: Messages emitted, which is not the same as executions: a node can run + #: and publish nothing. + messages: int = 0 + duration_sum_ms: float = 0.0 + duration_max_ms: float = 0.0 + #: How long queued work waited before it ran. + lag_sum_ms: float = 0.0 + lag_max_ms: float = 0.0 + #: Items the lag sums are over, so an average can be taken. + items: int = 0 + + +class EngineEvent(SQLModel, table=True): + """Something worth keeping after the websocket has forgotten it.""" + + __tablename__ = "engine_event" + + id: int | None = Field(default=None, primary_key=True) + ts: datetime = Field( + index=True, + sa_type=DateTime(timezone=True), # type: ignore + ) + #: node_error, flow_quarantined, engine_degraded, …, or audit. + type: str = Field(max_length=32, index=True) + flow: str = "" + node: str = "" + #: The error and its traceback, or what an audit entry says was done. + detail: str = "" + #: Who did it, on audit rows. + actor: str = "" + + +class FlowRun(SQLModel, table=True): + """One cascade, from the item that started it to the last node in it.""" + + __tablename__ = "flow_run" + + #: The queue entry id, or a `manual-` one for a run that never queued. + id: str = Field(primary_key=True, max_length=64) + flow: str = Field(index=True, max_length=255) + #: What caused it: the work item's cause, or "manual". + source: str = "" + started_at: datetime = Field( + index=True, + sa_type=DateTime(timezone=True), # type: ignore + ) + finished_at: datetime | None = Field( + default=None, + sa_type=DateTime(timezone=True), # type: ignore + ) + #: running, ok, error or abandoned. + status: str = "running" + nodes: int = 0 + errors: int = 0 + duration_ms: float = 0.0 + deliveries: int = 1 diff --git a/backend/tests/api/routes/test_observability.py b/backend/tests/api/routes/test_observability.py new file mode 100644 index 0000000..f1cdc98 --- /dev/null +++ b/backend/tests/api/routes/test_observability.py @@ -0,0 +1,104 @@ +from datetime import datetime, timedelta, timezone + +from fastapi.testclient import TestClient +from sqlmodel import Session + +from app.core.config import settings +from app.models import EngineEvent, FlowRun, MetricBucket + +PREFIX = f"{settings.API_V1_STR}/observability" +FLOW = "observability-test" + + +def _seed(db: Session) -> None: + now = datetime.now(timezone.utc).replace(second=0, microsecond=0) + db.add( + MetricBucket( + flow=FLOW, + node=f"{FLOW}.calc", + bucket=now - timedelta(minutes=1), + executions=4, + errors=1, + messages=6, + duration_sum_ms=40.0, + duration_max_ms=25.0, + lag_sum_ms=100.0, + lag_max_ms=60.0, + items=4, + ) + ) + db.add( + EngineEvent( + ts=now, + type="node_error", + flow=FLOW, + node=f"{FLOW}.calc", + detail="ValueError: bad input\nTraceback", + ) + ) + db.add( + EngineEvent( + ts=now, type="audit", flow=FLOW, detail="published", actor="a@example.com" + ) + ) + db.add( + FlowRun( + id="9-0", + flow=FLOW, + source="external", + started_at=now, + finished_at=now, + status="ok", + nodes=3, + duration_ms=12.5, + ) + ) + db.commit() + + +def test_observability_requires_authentication(client: TestClient) -> None: + assert client.get(f"{PREFIX}/summary").status_code == 401 + + +def test_the_summary_answers_even_when_degraded( + client: TestClient, superuser_token_headers: dict[str, str] +) -> None: + response = client.get(f"{PREFIX}/summary", headers=superuser_token_headers) + + assert response.status_code == 200 + body = response.json() + assert body["status"] in {"ok", "degraded"} + assert set(body["flows"]) == {"total", "running", "paused", "quarantined"} + assert "error" in body["nodes"] + + +def test_the_history_reads_back( + client: TestClient, superuser_token_headers: dict[str, str], db: Session +) -> None: + _seed(db) + + points = client.get(f"{PREFIX}/timeseries", headers=superuser_token_headers).json() + mine = [point for point in points if point["executions"]] + assert mine and mine[-1]["avg_ms"] > 0 + + flows = client.get(f"{PREFIX}/flows", headers=superuser_token_headers).json() + row = next(entry for entry in flows if entry["flow"] == FLOW) + assert (row["executions"], row["errors"], row["messages"]) == (4, 1, 6) + assert len(row["spark"]) == 60 + assert row["last_error_ts"] is not None + + runs = client.get(f"{PREFIX}/runs", headers=superuser_token_headers).json() + assert any(run["id"] == "9-0" and run["status"] == "ok" for run in runs) + + failures = client.get(f"{PREFIX}/events", headers=superuser_token_headers).json() + assert any("Traceback" in event["detail"] for event in failures) + assert all(event["type"] != "audit" for event in failures) + + audit = client.get( + f"{PREFIX}/events", headers=superuser_token_headers, params={"kind": "audit"} + ).json() + assert any(event["actor"] == "a@example.com" for event in audit) + + dead = client.get(f"{PREFIX}/dead-letter", headers=superuser_token_headers) + assert dead.status_code == 200 + assert isinstance(dead.json(), list) diff --git a/backend/tests/flow/test_alerts.py b/backend/tests/flow/test_alerts.py index aaca22c..cda1938 100644 --- a/backend/tests/flow/test_alerts.py +++ b/backend/tests/flow/test_alerts.py @@ -122,11 +122,11 @@ def test_a_flapping_connection_goes_quiet(): async def scenario(): for _ in range(FLAP_THRESHOLD * 2 + 6): await alerts.handle( - {"type": "node_health", "node": "heating.pump", "status": "down"} + {"type": "node_health", "node": "heating.pump", "health": "down"} ) clock.advance(5) await alerts.handle( - {"type": "node_health", "node": "heating.pump", "status": "ok"} + {"type": "node_health", "node": "heating.pump", "health": "ok"} ) clock.advance(5) @@ -185,7 +185,13 @@ def test_alerting_can_be_switched_off(): "The work queue is unreachable", ), ({"type": "engine_degraded", "reason": "lag"}, "The engine is struggling"), - ({"type": "node_health", "status": "ok"}, None), + ({"type": "node_health", "health": "ok"}, None), + # The engine publishes `health`, not `status`: reading the wrong key + # meant a device dropping never alerted anyone. + ( + {"type": "node_health", "node": "heating.pump", "health": "down"}, + "heating.pump lost its connection", + ), ], ) def test_every_alerting_event_reads_as_a_sentence(event, expected): diff --git a/backend/tests/test_metrics.py b/backend/tests/test_metrics.py new file mode 100644 index 0000000..425af0f --- /dev/null +++ b/backend/tests/test_metrics.py @@ -0,0 +1,149 @@ +"""The collector writes down what the bus only ever broadcast. + +Not under ``tests/flow`` with the rest of the engine: that package opts out of +the database, and writing to it is this module's whole job. +""" + +import asyncio +import time +from datetime import datetime, timezone + +from sqlmodel import Session, select + +from app.flow.events import EventBus +from app.flow.metrics import MetricsCollector +from app.models import EngineEvent, FlowRun, MetricBucket + +FLOW = "metrics-test" +NODE = "metrics-test.calc" + + +def _events(collector: MetricsCollector, ts: float, run: str) -> None: + collector.handle( + { + "type": "cascade_started", + "run": run, + "flow": FLOW, + "node": "metrics-test.in", + "cause": "external", + "deliveries": 1, + "ts": ts, + } + ) + collector.handle( + { + "type": "node_executed", + "flow": FLOW, + "node": NODE, + "outputs": 2, + "duration_ms": 5.0, + "run": run, + "ts": ts, + } + ) + collector.handle( + {"type": "work_latency", "flow": FLOW, "node": NODE, "lag_ms": 12.0, "ts": ts} + ) + # The traceback arrives as its own event, just before the failure. + collector.handle( + { + "type": "node_log", + "flow": FLOW, + "node": NODE, + "level": "error", + "text": "Traceback: line 3, in run", + "ts": ts, + } + ) + collector.handle( + { + "type": "node_error", + "flow": FLOW, + "node": NODE, + "error": "ValueError: bad input", + "run": run, + "ts": ts, + } + ) + + +def test_events_become_rollups_failures_runs_and_audit(db: Session) -> None: + collector = MetricsCollector(EventBus()) + # The current minute, pinned: the second flush has to land in the same + # bucket, and anything past the retention window is pruned on write. + ts = datetime.now(timezone.utc).replace(second=0, microsecond=0).timestamp() + _events(collector, ts, "1-0") + collector.handle( + { + "type": "audit", + "action": "published", + "flow": FLOW, + "user": "someone@example.com", + "ts": ts, + } + ) + collector.handle( + {"type": "cascade_finished", "run": "1-0", "flow": FLOW, "ts": ts + 0.5} + ) + asyncio.run(collector.flush()) + + bucket = db.exec( + select(MetricBucket).where(MetricBucket.flow == FLOW, MetricBucket.node == NODE) + ).one() + assert (bucket.executions, bucket.errors, bucket.messages) == (1, 1, 2) + assert bucket.duration_max_ms == 5.0 + assert (bucket.lag_max_ms, bucket.items) == (12.0, 1) + + failure = db.exec( + select(EngineEvent).where( + EngineEvent.flow == FLOW, EngineEvent.type == "node_error" + ) + ).one() + assert "ValueError: bad input" in failure.detail + assert "line 3, in run" in failure.detail + + audit = db.exec( + select(EngineEvent).where(EngineEvent.flow == FLOW, EngineEvent.type == "audit") + ).one() + assert (audit.actor, audit.detail) == ("someone@example.com", "published") + + run = db.exec(select(FlowRun).where(FlowRun.id == "1-0")).one() + assert (run.status, run.flow, run.source) == ("error", FLOW, "external") + assert (run.nodes, run.errors) == (1, 1) + assert run.duration_ms > 0 + + # The same minute, written again: the counters add rather than duplicate. + _events(collector, ts + 10, "2-0") + asyncio.run(collector.flush()) + db.expire_all() + bucket = db.exec( + select(MetricBucket).where(MetricBucket.flow == FLOW, MetricBucket.node == NODE) + ).one() + assert (bucket.executions, bucket.errors) == (2, 2) + + # Never finished, so it is still open — the prune is what closes it. + open_run = db.exec(select(FlowRun).where(FlowRun.id == "2-0")).one() + assert open_run.status == "running" + + +def test_a_run_that_did_not_fail_reads_ok(db: Session) -> None: + collector = MetricsCollector(EventBus()) + ts = time.time() + collector.handle( + { + "type": "cascade_started", + "run": "manual-abc", + "flow": FLOW, + "node": "metrics-test.in", + "cause": "manual", + "deliveries": 1, + "ts": ts, + } + ) + collector.handle( + {"type": "cascade_finished", "run": "manual-abc", "flow": FLOW, "ts": ts + 0.1} + ) + asyncio.run(collector.flush()) + + run = db.exec(select(FlowRun).where(FlowRun.id == "manual-abc")).one() + assert (run.status, run.source) == ("ok", "manual") diff --git a/frontend/src/client/schemas.gen.ts b/frontend/src/client/schemas.gen.ts index 19f8435..08507e7 100644 --- a/frontend/src/client/schemas.gen.ts +++ b/frontend/src/client/schemas.gen.ts @@ -379,6 +379,38 @@ export const DashboardsPublicSchema = { title: 'DashboardsPublic' } as const; +export const DeadLetterSchema = { + properties: { + id: { + type: 'string', + title: 'Id' + }, + ts: { + type: 'number', + title: 'Ts' + }, + flow: { + type: 'string', + title: 'Flow' + }, + node: { + type: 'string', + title: 'Node' + }, + cause: { + type: 'string', + title: 'Cause' + }, + reason: { + type: 'string', + title: 'Reason' + } + }, + type: 'object', + required: ['id', 'ts', 'flow', 'node', 'cause', 'reason'], + title: 'DeadLetter' +} as const; + export const EndpointSchema = { properties: { kind: { @@ -426,6 +458,43 @@ these so a value never appears to come from nowhere — or worse, appears to come from whichever node happens to be drawn as a producer.` } as const; +export const EventRowSchema = { + properties: { + id: { + type: 'integer', + title: 'Id' + }, + ts: { + type: 'string', + format: 'date-time', + title: 'Ts' + }, + type: { + type: 'string', + title: 'Type' + }, + flow: { + type: 'string', + title: 'Flow' + }, + node: { + type: 'string', + title: 'Node' + }, + detail: { + type: 'string', + title: 'Detail' + }, + actor: { + type: 'string', + title: 'Actor' + } + }, + type: 'object', + required: ['id', 'ts', 'type', 'flow', 'node', 'detail', 'actor'], + title: 'EventRow' +} as const; + export const FlowDef_InputSchema = { properties: { name: { @@ -597,6 +666,56 @@ export const FlowInput_OutputSchema = { description: 'A message the flow starts with rather than computes.' } as const; +export const FlowRollupSchema = { + properties: { + flow: { + type: 'string', + title: 'Flow' + }, + executions: { + type: 'integer', + title: 'Executions' + }, + errors: { + type: 'integer', + title: 'Errors' + }, + messages: { + type: 'integer', + title: 'Messages' + }, + avg_ms: { + type: 'number', + title: 'Avg Ms' + }, + avg_lag_ms: { + type: 'number', + title: 'Avg Lag Ms' + }, + spark: { + items: { + type: 'integer' + }, + type: 'array', + title: 'Spark' + }, + last_error_ts: { + anyOf: [ + { + type: 'number' + }, + { + type: 'null' + } + ], + title: 'Last Error Ts' + } + }, + type: 'object', + required: ['flow', 'executions', 'errors', 'messages', 'avg_ms', 'avg_lag_ms', 'spark'], + title: 'FlowRollup' +} as const; + export const FlowStatePublicSchema = { properties: { values: { @@ -698,6 +817,55 @@ export const HTTPValidationErrorSchema = { title: 'HTTPValidationError' } as const; +export const HealthSummarySchema = { + properties: { + status: { + type: 'string', + title: 'Status' + }, + problems: { + items: { + type: 'string' + }, + type: 'array', + title: 'Problems' + }, + flows: { + additionalProperties: { + type: 'integer' + }, + type: 'object', + title: 'Flows' + }, + nodes: { + additionalProperties: { + type: 'integer' + }, + type: 'object', + title: 'Nodes' + }, + queue: { + additionalProperties: true, + type: 'object', + title: 'Queue' + }, + loop_lag: { + additionalProperties: { + type: 'number' + }, + type: 'object', + title: 'Loop Lag' + }, + failures_24h: { + type: 'integer', + title: 'Failures 24H' + } + }, + type: 'object', + required: ['status', 'problems', 'flows', 'nodes', 'queue', 'loop_lag', 'failures_24h'], + title: 'HealthSummary' +} as const; + export const HistoryPointSchema = { properties: { ts: { @@ -1630,6 +1798,63 @@ export const RunRequestSchema = { title: 'RunRequest' } as const; +export const RunRowSchema = { + properties: { + id: { + type: 'string', + title: 'Id' + }, + flow: { + type: 'string', + title: 'Flow' + }, + source: { + type: 'string', + title: 'Source' + }, + status: { + type: 'string', + title: 'Status' + }, + started_at: { + type: 'string', + format: 'date-time', + title: 'Started At' + }, + finished_at: { + anyOf: [ + { + type: 'string', + format: 'date-time' + }, + { + type: 'null' + } + ], + title: 'Finished At' + }, + nodes: { + type: 'integer', + title: 'Nodes' + }, + errors: { + type: 'integer', + title: 'Errors' + }, + duration_ms: { + type: 'number', + title: 'Duration Ms' + }, + deliveries: { + type: 'integer', + title: 'Deliveries' + } + }, + type: 'object', + required: ['id', 'flow', 'source', 'status', 'started_at', 'nodes', 'errors', 'duration_ms', 'deliveries'], + title: 'RunRow' +} as const; + export const SecretNamesSchema = { properties: { data: { @@ -1711,6 +1936,42 @@ export const SectionDef_OutputSchema = { description: 'A grid of widgets under a heading.' } as const; +export const SeriesPointSchema = { + properties: { + ts: { + type: 'number', + title: 'Ts' + }, + executions: { + type: 'integer', + title: 'Executions' + }, + errors: { + type: 'integer', + title: 'Errors' + }, + messages: { + type: 'integer', + title: 'Messages' + }, + avg_ms: { + type: 'number', + title: 'Avg Ms' + }, + max_ms: { + type: 'number', + title: 'Max Ms' + }, + avg_lag_ms: { + type: 'number', + title: 'Avg Lag Ms' + } + }, + type: 'object', + required: ['ts', 'executions', 'errors', 'messages', 'avg_ms', 'max_ms', 'avg_lag_ms'], + title: 'SeriesPoint' +} as const; + export const ShareRequestSchema = { properties: { lib_name: { diff --git a/frontend/src/client/sdk.gen.ts b/frontend/src/client/sdk.gen.ts index 23a6ed8..8ad79b5 100644 --- a/frontend/src/client/sdk.gen.ts +++ b/frontend/src/client/sdk.gen.ts @@ -3,7 +3,7 @@ import type { CancelablePromise } from './core/CancelablePromise'; import { OpenAPI } from './core/OpenAPI'; import { request as __request } from './core/request'; -import type { AlertsReadAlertsConfigResponse, AlertsSaveAlertsConfigData, AlertsSaveAlertsConfigResponse, AlertsTestChannelData, AlertsTestChannelResponse, DashboardsReadDashboardsResponse, DashboardsReadDashboardData, DashboardsReadDashboardResponse, DashboardsCreateDashboardData, DashboardsCreateDashboardResponse, DashboardsSaveDashboardData, DashboardsSaveDashboardResponse, DashboardsDeleteDashboardData, DashboardsDeleteDashboardResponse, DashboardsRenameDashboardData, DashboardsRenameDashboardResponse, FlowsReadFlowsResponse, FlowsReadNodeTypesResponse, FlowsReadLibraryResponse, FlowsDeleteSharedNodeData, FlowsDeleteSharedNodeResponse, FlowsReadFlowData, FlowsReadFlowResponse, FlowsSaveFlowData, FlowsSaveFlowResponse, FlowsDeleteFlowData, FlowsDeleteFlowResponse, FlowsPublishFlowData, FlowsPublishFlowResponse, FlowsDiscardDraftData, FlowsDiscardDraftResponse, FlowsRenameFlowData, FlowsRenameFlowResponse, FlowsReadNodeSourceData, FlowsReadNodeSourceResponse, FlowsSaveNodeSourceData, FlowsSaveNodeSourceResponse, FlowsShareNodeData, FlowsShareNodeResponse, FlowsUnshareNodeData, FlowsUnshareNodeResponse, FlowsStartFlowData, FlowsStartFlowResponse, FlowsStopFlowData, FlowsStopFlowResponse, FlowsPauseFlowData, FlowsPauseFlowResponse, FlowsResumeFlowData, FlowsResumeFlowResponse, FlowsStepFlowData, FlowsStepFlowResponse, FlowsValidateFlowData, FlowsValidateFlowResponse, FlowsRunFlowData, FlowsRunFlowResponse, FlowsTriggerNodeData, FlowsTriggerNodeResponse, FlowsCancelNodeData, FlowsCancelNodeResponse, FlowsReadFlowStateData, FlowsReadFlowStateResponse, FlowsReadMessageHistoryData, FlowsReadMessageHistoryResponse, LoginLoginAccessTokenData, LoginLoginAccessTokenResponse, LoginTestTokenResponse, LoginRecoverPasswordData, LoginRecoverPasswordResponse, LoginResetPasswordData, LoginResetPasswordResponse, LoginRecoverPasswordHtmlContentData, LoginRecoverPasswordHtmlContentResponse, MessagesReadMessagesResponse, MessagesPublishMessageData, MessagesPublishMessageResponse, MessagesReadMessageHistoryData, MessagesReadMessageHistoryResponse, ModulesReadModulesResponse, ModulesApplyModulesData, ModulesApplyModulesResponse, OauthRegisterClientData, OauthRegisterClientResponse, OauthAuthorizeValidateData, OauthAuthorizeValidateResponse, OauthAuthorizeData, OauthAuthorizeResponse, OauthTokenData, OauthTokenResponse, OauthReadClientsResponse, OauthRevokeClientData, OauthRevokeClientResponse, PrivateCreateUserData, PrivateCreateUserResponse, SecretsReadSecretsResponse, SecretsSaveSecretData, SecretsSaveSecretResponse, SecretsDeleteSecretData, SecretsDeleteSecretResponse, UsersReadUsersData, UsersReadUsersResponse, UsersCreateUserData, UsersCreateUserResponse, UsersReadUserMeResponse, UsersDeleteUserMeResponse, UsersUpdateUserMeData, UsersUpdateUserMeResponse, UsersUpdatePasswordMeData, UsersUpdatePasswordMeResponse, UsersRegisterUserData, UsersRegisterUserResponse, UsersReadUserByIdData, UsersReadUserByIdResponse, UsersUpdateUserData, UsersUpdateUserResponse, UsersDeleteUserData, UsersDeleteUserResponse, UtilsTestEmailData, UtilsTestEmailResponse, UtilsHealthCheckResponse, UtilsHealthResponse } from './types.gen'; +import type { AlertsReadAlertsConfigResponse, AlertsSaveAlertsConfigData, AlertsSaveAlertsConfigResponse, AlertsTestChannelData, AlertsTestChannelResponse, DashboardsReadDashboardsResponse, DashboardsReadDashboardData, DashboardsReadDashboardResponse, DashboardsCreateDashboardData, DashboardsCreateDashboardResponse, DashboardsSaveDashboardData, DashboardsSaveDashboardResponse, DashboardsDeleteDashboardData, DashboardsDeleteDashboardResponse, DashboardsRenameDashboardData, DashboardsRenameDashboardResponse, FlowsReadFlowsResponse, FlowsReadNodeTypesResponse, FlowsReadLibraryResponse, FlowsDeleteSharedNodeData, FlowsDeleteSharedNodeResponse, FlowsReadFlowData, FlowsReadFlowResponse, FlowsSaveFlowData, FlowsSaveFlowResponse, FlowsDeleteFlowData, FlowsDeleteFlowResponse, FlowsPublishFlowData, FlowsPublishFlowResponse, FlowsDiscardDraftData, FlowsDiscardDraftResponse, FlowsRenameFlowData, FlowsRenameFlowResponse, FlowsReadNodeSourceData, FlowsReadNodeSourceResponse, FlowsSaveNodeSourceData, FlowsSaveNodeSourceResponse, FlowsShareNodeData, FlowsShareNodeResponse, FlowsUnshareNodeData, FlowsUnshareNodeResponse, FlowsStartFlowData, FlowsStartFlowResponse, FlowsStopFlowData, FlowsStopFlowResponse, FlowsPauseFlowData, FlowsPauseFlowResponse, FlowsResumeFlowData, FlowsResumeFlowResponse, FlowsStepFlowData, FlowsStepFlowResponse, FlowsValidateFlowData, FlowsValidateFlowResponse, FlowsRunFlowData, FlowsRunFlowResponse, FlowsTriggerNodeData, FlowsTriggerNodeResponse, FlowsCancelNodeData, FlowsCancelNodeResponse, FlowsReadFlowStateData, FlowsReadFlowStateResponse, FlowsReadMessageHistoryData, FlowsReadMessageHistoryResponse, LoginLoginAccessTokenData, LoginLoginAccessTokenResponse, LoginTestTokenResponse, LoginRecoverPasswordData, LoginRecoverPasswordResponse, LoginResetPasswordData, LoginResetPasswordResponse, LoginRecoverPasswordHtmlContentData, LoginRecoverPasswordHtmlContentResponse, MessagesReadMessagesResponse, MessagesPublishMessageData, MessagesPublishMessageResponse, MessagesReadMessageHistoryData, MessagesReadMessageHistoryResponse, ModulesReadModulesResponse, ModulesApplyModulesData, ModulesApplyModulesResponse, OauthRegisterClientData, OauthRegisterClientResponse, OauthAuthorizeValidateData, OauthAuthorizeValidateResponse, OauthAuthorizeData, OauthAuthorizeResponse, OauthTokenData, OauthTokenResponse, OauthReadClientsResponse, OauthRevokeClientData, OauthRevokeClientResponse, ObservabilityReadSummaryResponse, ObservabilityReadTimeseriesData, ObservabilityReadTimeseriesResponse, ObservabilityReadFlowRollupsData, ObservabilityReadFlowRollupsResponse, ObservabilityReadRunsData, ObservabilityReadRunsResponse, ObservabilityReadEventsData, ObservabilityReadEventsResponse, ObservabilityReadDeadLettersData, ObservabilityReadDeadLettersResponse, PrivateCreateUserData, PrivateCreateUserResponse, SecretsReadSecretsResponse, SecretsSaveSecretData, SecretsSaveSecretResponse, SecretsDeleteSecretData, SecretsDeleteSecretResponse, UsersReadUsersData, UsersReadUsersResponse, UsersCreateUserData, UsersCreateUserResponse, UsersReadUserMeResponse, UsersDeleteUserMeResponse, UsersUpdateUserMeData, UsersUpdateUserMeResponse, UsersUpdatePasswordMeData, UsersUpdatePasswordMeResponse, UsersRegisterUserData, UsersRegisterUserResponse, UsersReadUserByIdData, UsersReadUserByIdResponse, UsersUpdateUserData, UsersUpdateUserResponse, UsersDeleteUserData, UsersDeleteUserResponse, UtilsTestEmailData, UtilsTestEmailResponse, UtilsHealthCheckResponse, UtilsHealthResponse } from './types.gen'; export class AlertsService { /** @@ -1069,6 +1069,140 @@ export class OauthService { } } +export class ObservabilityService { + /** + * Read Summary + * How the engine is doing right now. Always 200, degraded or not. + * @returns HealthSummary Successful Response + * @throws ApiError + */ + public static readSummary(): CancelablePromise { + return __request(OpenAPI, { + method: 'GET', + url: '/api/v1/observability/summary' + }); + } + + /** + * Read Timeseries + * Executions, errors and timings over time, summed across nodes. + * @param data The data for the request. + * @param data.flow + * @param data.node + * @param data.hours + * @param data.bucketS + * @returns SeriesPoint Successful Response + * @throws ApiError + */ + public static readTimeseries(data: ObservabilityReadTimeseriesData = {}): CancelablePromise { + return __request(OpenAPI, { + method: 'GET', + url: '/api/v1/observability/timeseries', + query: { + flow: data.flow, + node: data.node, + hours: data.hours, + bucket_s: data.bucketS + }, + errors: { + 422: 'Validation Error' + } + }); + } + + /** + * Read Flow Rollups + * One row per flow, with a coarse trend of how much it ran. + * @param data The data for the request. + * @param data.hours + * @returns FlowRollup Successful Response + * @throws ApiError + */ + public static readFlowRollups(data: ObservabilityReadFlowRollupsData = {}): CancelablePromise { + return __request(OpenAPI, { + method: 'GET', + url: '/api/v1/observability/flows', + query: { + hours: data.hours + }, + errors: { + 422: 'Validation Error' + } + }); + } + + /** + * Read Runs + * Recent cascades, newest first. + * @param data The data for the request. + * @param data.flow + * @param data.status + * @param data.limit + * @returns RunRow Successful Response + * @throws ApiError + */ + public static readRuns(data: ObservabilityReadRunsData = {}): CancelablePromise { + return __request(OpenAPI, { + method: 'GET', + url: '/api/v1/observability/runs', + query: { + flow: data.flow, + status: data.status, + limit: data.limit + }, + errors: { + 422: 'Validation Error' + } + }); + } + + /** + * Read Events + * What went wrong, or who changed what. Newest first. + * @param data The data for the request. + * @param data.kind + * @param data.flow + * @param data.limit + * @returns EventRow Successful Response + * @throws ApiError + */ + public static readEvents(data: ObservabilityReadEventsData = {}): CancelablePromise { + return __request(OpenAPI, { + method: 'GET', + url: '/api/v1/observability/events', + query: { + kind: data.kind, + flow: data.flow, + limit: data.limit + }, + errors: { + 422: 'Validation Error' + } + }); + } + + /** + * Read Dead Letters + * Work the engine gave up on, which nothing else surfaces. + * @param data The data for the request. + * @param data.limit + * @returns DeadLetter Successful Response + * @throws ApiError + */ + public static readDeadLetters(data: ObservabilityReadDeadLettersData = {}): CancelablePromise { + return __request(OpenAPI, { + method: 'GET', + url: '/api/v1/observability/dead-letter', + query: { + limit: data.limit + }, + errors: { + 422: 'Validation Error' + } + }); + } +} + export class PrivateService { /** * Create User diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts index 67ee1a0..c110dac 100644 --- a/frontend/src/client/types.gen.ts +++ b/frontend/src/client/types.gen.ts @@ -129,6 +129,15 @@ export type DashboardSummary = { widget_count?: number; }; +export type DeadLetter = { + id: string; + ts: number; + flow: string; + node: string; + cause: string; + reason: string; +}; + /** * Serializable payload types. * @@ -154,6 +163,16 @@ export type Endpoint = { requires?: Array<(string)>; }; +export type EventRow = { + id: number; + ts: string; + type: string; + flow: string; + node: string; + detail: string; + actor: string; +}; + /** * One atomic flow. */ @@ -209,6 +228,17 @@ export type FlowInput_Output = { initial?: (unknown | null); }; +export type FlowRollup = { + flow: string; + executions: number; + errors: number; + messages: number; + avg_ms: number; + avg_lag_ms: number; + spark: Array<(number)>; + last_error_ts?: (number | null); +}; + export type FlowsPublic = { data: Array; count: number; @@ -232,6 +262,24 @@ export type FlowSummary = { quarantined?: boolean; }; +export type HealthSummary = { + status: string; + problems: Array<(string)>; + flows: { + [key: string]: (number); + }; + nodes: { + [key: string]: (number); + }; + queue: { + [key: string]: unknown; + }; + loop_lag: { + [key: string]: (number); + }; + failures_24h: number; +}; + /** * One numeric value a message carried, and when. */ @@ -530,6 +578,19 @@ export type RunRequest = { }; }; +export type RunRow = { + id: string; + flow: string; + source: string; + status: string; + started_at: string; + finished_at?: (string | null); + nodes: number; + errors: number; + duration_ms: number; + deliveries: number; +}; + export type SecretNames = { data: Array<(string)>; count: number; @@ -557,6 +618,16 @@ export type SectionDef_Output = { widgets?: Array; }; +export type SeriesPoint = { + ts: number; + executions: number; + errors: number; + messages: number; + avg_ms: number; + max_ms: number; + avg_lag_ms: number; +}; + export type ShareRequest = { lib_name: string; }; @@ -946,6 +1017,45 @@ export type OauthRevokeClientData = { export type OauthRevokeClientResponse = (Message); +export type ObservabilityReadSummaryResponse = (HealthSummary); + +export type ObservabilityReadTimeseriesData = { + bucketS?: number; + flow?: (string | null); + hours?: number; + node?: (string | null); +}; + +export type ObservabilityReadTimeseriesResponse = (Array); + +export type ObservabilityReadFlowRollupsData = { + hours?: number; +}; + +export type ObservabilityReadFlowRollupsResponse = (Array); + +export type ObservabilityReadRunsData = { + flow?: (string | null); + limit?: number; + status?: (string | null); +}; + +export type ObservabilityReadRunsResponse = (Array); + +export type ObservabilityReadEventsData = { + flow?: (string | null); + kind?: 'failure' | 'audit'; + limit?: number; +}; + +export type ObservabilityReadEventsResponse = (Array); + +export type ObservabilityReadDeadLettersData = { + limit?: number; +}; + +export type ObservabilityReadDeadLettersResponse = (Array); + export type PrivateCreateUserData = { requestBody: PrivateUserCreate; }; diff --git a/frontend/src/components/Common/UplotChart.tsx b/frontend/src/components/Common/UplotChart.tsx new file mode 100644 index 0000000..96564d0 --- /dev/null +++ b/frontend/src/components/Common/UplotChart.tsx @@ -0,0 +1,163 @@ +import { useEffect, useLayoutEffect, useRef } from "react" +import uPlot from "uplot" +import "uplot/dist/uPlot.min.css" + +import type { HistoryPoint } from "@/client" +import { useTheme } from "@/components/theme-provider" + +/** + * How many lines one chart carries. + * + * The bound is the palette's: `--chart-1…5` is one designed ramp, and a sixth + * line would either repeat a step or invent a colour outside the system. + */ +export const MAX_SERIES = 5 + +/** Room for the axis ticks; uPlot measures the rest of the box itself. */ +const PADDING: uPlot.Padding = [10, 12, 0, 0] + +/** The legend sits under the canvas, so the canvas has to leave it room. */ +const LEGEND_HEIGHT = 26 + +const canvasHeight = (element: HTMLElement) => + Math.max(60, (element.clientHeight || 180) - LEGEND_HEIGHT) + +/** A token, resolved for the canvas — which cannot read CSS variables. */ +function token(name: string): string { + return getComputedStyle(document.documentElement) + .getPropertyValue(name) + .trim() +} + +const seriesColor = (index: number) => token(`--chart-${(index % 5) + 1}`) + +/** The series joined onto one x axis, which is what uPlot draws. */ +function table(plots: HistoryPoint[][]): uPlot.AlignedData { + return uPlot.join( + plots.map( + (plot) => + [ + plot.map((point) => point.ts), + plot.map((point) => point.value), + ] as uPlot.AlignedData, + ), + ) +} + +/** + * Several series over time, drawn on one axis. + * + * uPlot rather than SVG: a chart may hold five series of hundreds of points + * each, which is more path data than React should be rebuilding on every value + * that arrives. Its own legend doubles as the hover readout, so the cursor + * tells you what each line was worth at that moment — and with more than one + * line a legend is required anyway. + */ +export function UplotChart({ + labels, + plots, + empty = "Nothing has come through yet.", +}: { + /** One label per series; the set of them is the chart's identity. */ + labels: string[] + /** The points of each series, in the same order as `labels`. */ + plots: HistoryPoint[][] + empty?: string +}) { + const host = useRef(null) + const chart = useRef(null) + const { resolvedTheme } = useTheme() + + const points = plots.reduce((total, plot) => total + plot.length, 0) + // The identity of the series set: the chart is rebuilt when it changes, + // while a new reading only sets its data. + const key = labels.join(" ") + // uPlot leaves its axes half-initialised while the scales have no range, and + // a resize in that window (a card still settling, say) draws them anyway and + // throws. Waiting for the first reading avoids the state altogether. + const ready = points > 0 + + // biome-ignore lint/correctness/useExhaustiveDependencies: the label string is the identity of the series set. + useLayoutEffect(() => { + const element = host.current + if (!element || labels.length === 0 || !ready) return + + const axis = { + stroke: () => token("--muted-foreground"), + grid: { stroke: () => token("--border"), width: 1 }, + ticks: { stroke: () => token("--border"), width: 1 }, + font: `11px ${getComputedStyle(element).fontFamily}`, + } + + const plot = new uPlot( + { + width: element.clientWidth || 320, + height: canvasHeight(element), + padding: PADDING, + cursor: { y: false }, + legend: { live: true }, + scales: { x: { time: true } }, + axes: [ + { ...axis, size: 28 }, + { ...axis, size: 46 }, + ], + series: [ + {}, + ...labels.map((label, index) => ({ + label, + width: 2, + // Read at draw time, so a theme toggle is a redraw rather than a + // rebuilt chart. + stroke: () => seriesColor(index), + // Series arrive on their own clocks; a joined table is mostly + // holes, and a line with a hole per point is not a line. + spanGaps: true, + points: { show: false }, + })), + ], + }, + // Built with the readings it already has: uPlot's axes are only half + // initialised while its scales have no range. + table(plots), + element, + ) + chart.current = plot + + const observer = new ResizeObserver(() => { + plot.setSize({ + width: element.clientWidth, + height: canvasHeight(element), + }) + }) + observer.observe(element) + + return () => { + observer.disconnect() + plot.destroy() + chart.current = null + } + }, [key, ready]) + + // biome-ignore lint/correctness/useExhaustiveDependencies: rebuilding the joined table is what the point count stands for. + useEffect(() => { + if (!chart.current || plots.length === 0) return + chart.current.setData(table(plots)) + }, [points, key]) + + // The canvas cannot follow a CSS variable, so a theme swap is a redraw. + // biome-ignore lint/correctness/useExhaustiveDependencies: the theme is the signal, not something the effect reads. + useEffect(() => { + chart.current?.redraw() + }, [resolvedTheme]) + + return ( +
+
+ {points === 0 ? ( +

+ {empty} +

+ ) : null} +
+ ) +} diff --git a/frontend/src/components/Dashboard/ChartWidget.tsx b/frontend/src/components/Dashboard/ChartWidget.tsx index 4a245fa4c7e9318d6a3dc38fbace25b05039ee89..3da9b75552e55ccbd41f5458b08a5c0da1e3710e 100644 GIT binary patch delta 267 zcmZ2yb4GT8*TiJeiCfIYd>td=gI$9>U4wNLLJM;8OPn(ji%KTzF^UN|=jZ0;=jp>m zC*F6T9Ltzmudfi2k*biES(d6>oLZEbTC9+qnwgWCm#&amtWW}!N(LFBkda?pqF$_{ zkeHVO7A;83OU+RzN=-~DR>%ZOrxvB^aivxifKFe5zcR&qD(kRb3!(y@dAUEMwUnNNWd(Lq}T}vVf5I1B-`pOy@;e~0zO2} zkc0Uo`Ks!5v#C)gfh}N$&3%LD}Qgugxm0%GQf3JC%X!(v-H@RJLk1sV<|mbjMX0o|}=5 zd>u@}q8!L9KMoT=D2mQreS>n9S={^_Q__5OnTAQ(9P`Z}h(eoe_Y(v{(!X#tvAb@RM+yhM?ZvvPI~D%~sIZVt+HC1$9tivuiq{FGOXcheeo7pf=n~G&RZ1qJYOH5zIy*E?nFfG=%^$ zyuqhugFW5v^z6HnzrDVBfBEj>?G5kw+5W!ndX*w>cm-?LELp{2a9?~)>gTaFMU_Je z9>%BKnxDt%5(gGGnzoL)JUKnRc>8Loi~UP8nS{v<$Bhp5dvx%;NBG#k-O3Z$nT2r* z>;x&`igHe1k_Q91pjJLgztY^`7AbUWpj*hA!kSgmw&-m!eGu~B|sAKjhENgaF;wZV^~sznpR1` zai{xviHYqH?wLcM-THynxN>gGDo+Rz_Y$yBm3DGnt|IGA(x757{raJn6l;il7oj#U zR{}3`LRZIdO6+y_Hnjo^3G9~$g`D*VAGFIMB~?7OIV;ZHu?fac*E)y(NMCg6FLbcE zbLzQ)IWtDRqVZwCbY-GwMU15gX|_0{BUG1}0zi!y&;s5{P?aW(b`=!cd}h6=ErYq8 zz^E;39o=F%7JT?NFxgfDR-?vA5hj6!jBdA)`}8B5%n-eOI=DUdarxt8+TZGaN5=dLq}!R=y`gq4r> zoZVfUMTiRe4711O8h+<#{reM&_+BrKO71F@SHr->tp5bdOIAI9* zzne{Y0Pt3eqRW_+9<>;%#~4H8m=jK7fmKd^6fvFJ&h=((?xo-?`rOD-aG#SclNc(G zs51(JFaV+gK*I#_AEjxgjwYec8J?4>4dKSKtv?C6B2&vfnH(cVx0#_B?cX|kJ~4Li zk+J6=8T;%bW1s67$bv-64_oAzI6UM;kp}k&SA}gB%NdOpFmesvbdqI=K4v0g&^#A@ zxs`E#%L^9(#|9S#wFGdXKSizxB$Bb#IqZlro+*kgZ}{m}tu2Ma;gt>2d~&GSy7wp} zy?EgvI-8B|?FviP9ezY+^g4g&Y)4Vc#9+}!lKa$Mb=Tq=sz*7>yZ8ZZk9J#{{mJ7R$%{`;DWK?)n zGPC*DbjN-p#2eHWqz9mz=g2jxLPb4R8=ohYm*>glH6Cj9Ax-A?Axx{HHCXF9u}hG( zk1G`_T5oU+|AB=wKgRtgcs&I@w{Aq$3`TZ_WZ_9v^$;YWGdsuD2TR~P<&R|At=p)c zac?uGl=bCDe84RO@aXYOq&m!F>cbXmwh5-fNv;nAhQAyCh)}?#SH;%%YPGS-)mN1GHR2zOu9~lq(7CO=l51~z1QV

i z$%AOE9CE#N8a}u}Xn93k$D&FelBNQwn7vNFA9MBE2Rz#u z_;fPqIL1LV%R`iQptDae93$PLjnze|5>CqbkPc{5ySTJ7e&gC{%58$XgM(~IpRPYB zv7H*aKF|Bk#9_4B-08Wfo!QhXJ%UX2-3vIC*kMhDSNEgA4$b+v5eO zC6k0Tp#=c22T!OcwWpaSPf{|;iXhBIj$l6kGKN24(etm`3!a$p`Iqep4Vb}wBo92= zzA-o-hNCoM{UVFPQb?7CtGsT9t7GQ8ivM)n`Ki8gKIKXaediDc-wyFt*LR2h_k);3~OKSa5#4g{JdQ z_MLmVgQ^gcj|Ki-V@k{8oNGN1_w~L0&wzJ~$$iu}^W8&(4|;vahT24&s9A&GF92H! z*b!2xdV<^@O?9B}s(JZXJ)EdzK66B7K^lR|ec(pFIm>BaNd3mi?6K2IZNdZUg#j`L zDnk4aG?7-g%Xm)|ACk&;r2yEnti{s_jg_H@E)lckN6g}!sytrL-sis44hV`Oh4aZ} z#Y^Q0H9^b8oxeOeQP^~Au6p<-`_~6<{e&O06SCiX8VN#0RGKc&X1i%X_7n<>#(p9?j)0vJNwVtSO4OAofR{ z66ICDh|#db=La0;#U*}kfeu~w$3Glo>^njr>Od7=I-G`s@EtgUG0G^EOp8@-II4t&>fz{neEcSe@k*#b3I_y}KyV>husl L!9K7&+*AJ#S^$DJ diff --git a/frontend/src/components/Flow/MessageSparkline.tsx b/frontend/src/components/Flow/MessageSparkline.tsx index f9e4a27..6e971df 100644 --- a/frontend/src/components/Flow/MessageSparkline.tsx +++ b/frontend/src/components/Flow/MessageSparkline.tsx @@ -49,7 +49,7 @@ function useSettled(value: string): string { * that never moved has no span to divide by, and one spanning decades is only * legible once the exponent is what varies. */ -function shape(points: HistoryPoint[]) { +export function shape(points: HistoryPoint[]) { const values = points.map((point) => point.value) const low = Math.min(...values) const high = Math.max(...values) diff --git a/frontend/src/components/Flow/liveStore.ts b/frontend/src/components/Flow/liveStore.ts index a33ab24..d057301 100644 --- a/frontend/src/components/Flow/liveStore.ts +++ b/frontend/src/components/Flow/liveStore.ts @@ -24,6 +24,19 @@ export type LiveStatus = { status: "active" | "error" | "running" | "success" error?: string | null } +/** How a node's connection is doing, which is not how its last run went. */ +export type NodeHealth = { + health: "ok" | "down" | "unknown" + detail?: string | null +} +/** Something the engine reported about itself, for the health page. */ +export type EngineEvent = { + type: string + flow?: string + node?: string + detail?: string + ts: number +} /** One node execution's output, as the log panel shows it. */ export type LogLine = { flow: string @@ -38,9 +51,13 @@ type Listener = () => void /** Enough to see what a flow has been doing, not a log store. */ const LOG_LIMIT = 500 +/** The health page reads these to know when to refetch; it is not a history. */ +const ENGINE_EVENT_LIMIT = 100 const values = new Map() const statuses = new Map() +const health = new Map() +let engineEvents: EngineEvent[] = [] // How many times a node has emitted. The number itself means nothing; a change // is what restarts the pulse. const emits = new Map() @@ -100,6 +117,18 @@ export const liveStore = { getStatus(nodeId: string) { return statuses.get(nodeId) }, + setHealth(nodeId: string, entry: NodeHealth) { + health.set(nodeId, entry) + notify(`health:${nodeId}`) + }, + getHealth(nodeId: string) { + return health.get(nodeId) + }, + recordEngineEvent(event: EngineEvent) { + // A new array each time, so the hook's snapshot comparison sees the change. + engineEvents = [...engineEvents, event].slice(-ENGINE_EVENT_LIMIT) + notify("engine") + }, recordEmit(nodeId: string) { emits.set(nodeId, (emits.get(nodeId) ?? 0) + 1) notify(`emit:${nodeId}`) @@ -146,6 +175,10 @@ export const liveStore = { statuses.clear() for (const key of emits.keys()) notify(`emit:${key}`) emits.clear() + for (const key of health.keys()) notify(`health:${key}`) + health.clear() + engineEvents = [] + notify("engine") logLines = [] notify("logs") for (const flow of paused) notify(`paused:${flow}`) @@ -167,6 +200,22 @@ export function useNodeStatus(nodeId: string): LiveStatus | undefined { ) } +/** How the node's connection is doing, once it has said anything about it. */ +export function useNodeHealth(nodeId: string): NodeHealth | undefined { + return useSyncExternalStore( + (listener) => subscribeKey(`health:${nodeId}`, listener), + () => health.get(nodeId), + ) +} + +/** The last hundred things the engine said about itself, oldest first. */ +export function useEngineEvents(): EngineEvent[] { + return useSyncExternalStore( + (listener) => subscribeKey("engine", listener), + () => engineEvents, + ) +} + /** Increments each time the node publishes something. */ export function useNodeEmits(nodeId: string): number { return useSyncExternalStore( diff --git a/frontend/src/components/Flow/useFlowSocket.ts b/frontend/src/components/Flow/useFlowSocket.ts index fed7050..c823298 100644 --- a/frontend/src/components/Flow/useFlowSocket.ts +++ b/frontend/src/components/Flow/useFlowSocket.ts @@ -27,11 +27,48 @@ type FlowEvent = source?: ValueSource } | { type: "node_started"; node: string } - | { type: "node_executed"; node: string; outputs: number } - | { type: "node_error"; node: string; error: string } + | { + type: "node_executed" + flow?: string + node: string + outputs: number + duration_ms?: number + } + | { + type: "node_error" + flow?: string + node: string + error: string + ts?: number + } | { type: "node_status"; node: string; status: string; error?: string | null } | ({ type: "node_log" } & LogLine) | { type: "flow_paused"; flow: string; paused: boolean } + | { + type: "node_health" + flow?: string + node: string + health: "ok" | "down" | "unknown" + detail?: string | null + ts?: number + } + | { type: "flow_quarantined"; flow: string; error?: string; ts?: number } + | { type: "engine_degraded"; reason?: string; ts?: number } + | { type: "engine_fatal"; reason?: string; ts?: number } + | { + type: "cascade_dropped" + flow?: string + node?: string + deliveries?: number + ts?: number + } + | { + type: "queue_unavailable" + flow?: string + node?: string + error?: string + ts?: number + } | { type: "pipeline_rebuilt" nodes: { id: string; status: string; error?: string | null }[] @@ -110,6 +147,44 @@ export function useFlowSocket(onAuthFailure?: () => void): void { status: "error", error: message.error, }) + liveStore.recordEngineEvent({ + type: message.type, + flow: message.flow, + node: message.node, + detail: message.error, + ts: message.ts ?? Date.now() / 1000, + }) + break + case "node_health": + liveStore.setHealth(message.node, { + health: message.health, + detail: message.detail, + }) + if (message.health === "down") { + liveStore.recordEngineEvent({ + type: message.type, + flow: message.flow, + node: message.node, + detail: message.detail ?? "Reported itself down.", + ts: message.ts ?? Date.now() / 1000, + }) + } + break + case "flow_quarantined": + case "engine_degraded": + case "engine_fatal": + case "cascade_dropped": + case "queue_unavailable": + liveStore.recordEngineEvent({ + type: message.type, + flow: "flow" in message ? message.flow : undefined, + node: "node" in message ? message.node : undefined, + detail: + ("error" in message ? message.error : undefined) ?? + ("reason" in message ? message.reason : undefined) ?? + "", + ts: message.ts ?? Date.now() / 1000, + }) break case "node_status": liveStore.setStatus(message.node, { diff --git a/frontend/src/components/Sidebar/AppSidebar.tsx b/frontend/src/components/Sidebar/AppSidebar.tsx index ac38a46..de843da 100644 --- a/frontend/src/components/Sidebar/AppSidebar.tsx +++ b/frontend/src/components/Sidebar/AppSidebar.tsx @@ -1,4 +1,5 @@ import { + Activity, Bell, Home, KeyRound, @@ -26,6 +27,7 @@ const baseItems: Item[] = [ { icon: Home, title: "Home", path: "/" }, { icon: Workflow, title: "Flows", path: "/flows" }, { icon: LayoutDashboard, title: "Dashboards", path: "/dashboards" }, + { icon: Activity, title: "Health", path: "/health" }, // Both are engine-wide operator settings rather than personal ones, so they // sit here and not among the per-user tabs under Settings. { icon: KeyRound, title: "Secrets", path: "/secrets" }, diff --git a/frontend/src/routeTree.gen.ts b/frontend/src/routeTree.gen.ts index 478a5b3..794b3e9 100644 --- a/frontend/src/routeTree.gen.ts +++ b/frontend/src/routeTree.gen.ts @@ -21,6 +21,7 @@ import { Route as OauthAuthorizeRouteImport } from './routes/oauth.authorize' import { Route as LayoutSettingsRouteImport } from './routes/_layout/settings' import { Route as LayoutSecretsRouteImport } from './routes/_layout/secrets' import { Route as LayoutModulesRouteImport } from './routes/_layout/modules' +import { Route as LayoutHealthRouteImport } from './routes/_layout/health' import { Route as LayoutAlertsRouteImport } from './routes/_layout/alerts' import { Route as LayoutAdminRouteImport } from './routes/_layout/admin' import { Route as LayoutFlowsIndexRouteImport } from './routes/_layout/flows/index' @@ -86,6 +87,11 @@ const LayoutModulesRoute = LayoutModulesRouteImport.update({ path: '/modules', getParentRoute: () => LayoutRoute, } as any) +const LayoutHealthRoute = LayoutHealthRouteImport.update({ + id: '/health', + path: '/health', + getParentRoute: () => LayoutRoute, +} as any) const LayoutAlertsRoute = LayoutAlertsRouteImport.update({ id: '/alerts', path: '/alerts', @@ -125,6 +131,7 @@ export interface FileRoutesByFullPath { '/signup': typeof SignupRoute '/admin': typeof LayoutAdminRoute '/alerts': typeof LayoutAlertsRoute + '/health': typeof LayoutHealthRoute '/modules': typeof LayoutModulesRoute '/secrets': typeof LayoutSecretsRoute '/settings': typeof LayoutSettingsRoute @@ -143,6 +150,7 @@ export interface FileRoutesByTo { '/signup': typeof SignupRoute '/admin': typeof LayoutAdminRoute '/alerts': typeof LayoutAlertsRoute + '/health': typeof LayoutHealthRoute '/modules': typeof LayoutModulesRoute '/secrets': typeof LayoutSecretsRoute '/settings': typeof LayoutSettingsRoute @@ -163,6 +171,7 @@ export interface FileRoutesById { '/signup': typeof SignupRoute '/_layout/admin': typeof LayoutAdminRoute '/_layout/alerts': typeof LayoutAlertsRoute + '/_layout/health': typeof LayoutHealthRoute '/_layout/modules': typeof LayoutModulesRoute '/_layout/secrets': typeof LayoutSecretsRoute '/_layout/settings': typeof LayoutSettingsRoute @@ -184,6 +193,7 @@ export interface FileRouteTypes { | '/signup' | '/admin' | '/alerts' + | '/health' | '/modules' | '/secrets' | '/settings' @@ -202,6 +212,7 @@ export interface FileRouteTypes { | '/signup' | '/admin' | '/alerts' + | '/health' | '/modules' | '/secrets' | '/settings' @@ -221,6 +232,7 @@ export interface FileRouteTypes { | '/signup' | '/_layout/admin' | '/_layout/alerts' + | '/_layout/health' | '/_layout/modules' | '/_layout/secrets' | '/_layout/settings' @@ -330,6 +342,13 @@ declare module '@tanstack/react-router' { preLoaderRoute: typeof LayoutModulesRouteImport parentRoute: typeof LayoutRoute } + '/_layout/health': { + id: '/_layout/health' + path: '/health' + fullPath: '/health' + preLoaderRoute: typeof LayoutHealthRouteImport + parentRoute: typeof LayoutRoute + } '/_layout/alerts': { id: '/_layout/alerts' path: '/alerts' @@ -391,6 +410,7 @@ const CanvasRouteWithChildren = interface LayoutRouteChildren { LayoutAdminRoute: typeof LayoutAdminRoute LayoutAlertsRoute: typeof LayoutAlertsRoute + LayoutHealthRoute: typeof LayoutHealthRoute LayoutModulesRoute: typeof LayoutModulesRoute LayoutSecretsRoute: typeof LayoutSecretsRoute LayoutSettingsRoute: typeof LayoutSettingsRoute @@ -402,6 +422,7 @@ interface LayoutRouteChildren { const LayoutRouteChildren: LayoutRouteChildren = { LayoutAdminRoute: LayoutAdminRoute, LayoutAlertsRoute: LayoutAlertsRoute, + LayoutHealthRoute: LayoutHealthRoute, LayoutModulesRoute: LayoutModulesRoute, LayoutSecretsRoute: LayoutSecretsRoute, LayoutSettingsRoute: LayoutSettingsRoute, diff --git a/frontend/src/routes/_layout/health.tsx b/frontend/src/routes/_layout/health.tsx new file mode 100644 index 0000000..45a1941 --- /dev/null +++ b/frontend/src/routes/_layout/health.tsx @@ -0,0 +1,457 @@ +import { useQuery, useQueryClient } from "@tanstack/react-query" +import { createFileRoute, Link } from "@tanstack/react-router" +import { ChevronDown, ChevronRight } from "lucide-react" +import { useEffect, useState } from "react" + +import { + type EventRow, + type FlowRollup, + type HistoryPoint, + ObservabilityService, +} from "@/client" +import { UplotChart } from "@/components/Common/UplotChart" +import { useEngineEvents } from "@/components/Flow/liveStore" +import { shape } from "@/components/Flow/MessageSparkline" +import { PANEL_SECTION } from "@/components/Flow/SidePanel" +import { useFlowSocket } from "@/components/Flow/useFlowSocket" +import { Badge } from "@/components/ui/badge" + +export const Route = createFileRoute("/_layout/health")({ + component: Health, + head: () => ({ + meta: [ + { + title: "Health - Fluksio", + }, + ], + }), +}) + +const healthKeys = { + all: ["observability"] as const, + summary: ["observability", "summary"] as const, + events: ["observability", "events"] as const, +} + +/** The live view; anything older is the collector's rollups. */ +const HOURS = 24 +const CARD = "rounded-lg border border-border bg-card p-4 shadow-e1" + +function ago(ts: string | number | null | undefined): string { + if (!ts) return "—" + const stamp = typeof ts === "number" ? ts * 1000 : Date.parse(ts) + const seconds = Math.max(0, (Date.now() - stamp) / 1000) + if (seconds < 90) return `${Math.round(seconds)}s ago` + if (seconds < 5400) return `${Math.round(seconds / 60)}m ago` + if (seconds < 172800) return `${Math.round(seconds / 3600)}h ago` + return `${Math.round(seconds / 86400)}d ago` +} + +const round = (value: number) => + value >= 100 ? Math.round(value) : +value.toFixed(1) + +function Tile({ + label, + value, + note, +}: { + label: string + value: string + note?: string +}) { + return ( +

+
{label}
+
{value}
+ {note ? ( +
{note}
+ ) : null} +
+ ) +} + +/** A flow's execution trend, drawn from the 60 slices the rollup carries. */ +function Spark({ counts }: { counts: number[] }) { + const points: HistoryPoint[] = counts.map((value, index) => ({ + ts: index, + value, + })) + if (points.every((point) => point.value === 0)) { + return nothing yet + } + const { line } = shape(points) + return ( + + ) +} + +function Failure({ event }: { event: EventRow }) { + const [open, setOpen] = useState(false) + const [first, ...rest] = event.detail.split("\n") + return ( +
+ + {open && rest.length ? ( + 
+          {rest.join("\n")}
+
+ ) : null} +
+ ) +} + +function Health() { + // The page is a socket subscriber like the editor: a failure should appear + // without waiting for the next poll. + useFlowSocket() + const live = useEngineEvents() + const queryClient = useQueryClient() + + const { data: summary } = useQuery({ + queryKey: healthKeys.summary, + queryFn: () => ObservabilityService.readSummary(), + refetchInterval: 10000, + }) + const { data: series } = useQuery({ + queryKey: ["observability", "timeseries", HOURS], + queryFn: () => ObservabilityService.readTimeseries({ hours: HOURS }), + refetchInterval: 30000, + }) + const { data: flows } = useQuery({ + queryKey: ["observability", "flows", HOURS], + queryFn: () => ObservabilityService.readFlowRollups({ hours: HOURS }), + refetchInterval: 30000, + }) + const { data: runs } = useQuery({ + queryKey: ["observability", "runs"], + queryFn: () => ObservabilityService.readRuns({ limit: 15 }), + refetchInterval: 30000, + }) + const { data: failures } = useQuery({ + queryKey: [...healthKeys.events, "failure"], + queryFn: () => + ObservabilityService.readEvents({ kind: "failure", limit: 25 }), + refetchInterval: 30000, + }) + const { data: audit } = useQuery({ + queryKey: [...healthKeys.events, "audit"], + queryFn: () => + ObservabilityService.readEvents({ kind: "audit", limit: 15 }), + refetchInterval: 30000, + }) + const { data: dead } = useQuery({ + queryKey: ["observability", "dead-letter"], + queryFn: () => ObservabilityService.readDeadLetters({ limit: 20 }), + refetchInterval: 30000, + }) + + // Something just went wrong on the socket. The row for it is written on the + // collector's next flush, so the refetch waits that out rather than asking + // for a failure the database does not have yet. + const seen = live.length + useEffect(() => { + if (!seen) return + const timer = setTimeout(() => { + queryClient.invalidateQueries({ queryKey: healthKeys.events }) + queryClient.invalidateQueries({ queryKey: healthKeys.summary }) + }, 16000) + return () => clearTimeout(timer) + }, [seen, queryClient]) + + const points = series ?? [] + const at = ( + pick: (point: (typeof points)[number]) => number, + ): HistoryPoint[] => + points.map((point) => ({ ts: point.ts, value: pick(point) })) + + const queue = (summary?.queue ?? {}) as Record + const degraded = summary?.status === "degraded" + + return ( +
+
+
+

Health

+ + {degraded ? "Degraded" : "Running normally"} + +
+

+ {summary?.problems.length + ? summary.problems.join(" · ") + : "What the engine has been doing over the last day, and what it is doing now."} +

+
+ +
+ + + + + +
+ +
+
+

Throughput per minute

+ point.messages), + at((point) => point.executions), + ]} + empty="Nothing has run yet." + /> +
+
+

Failures and timing

+ point.errors), + at((point) => point.avg_ms), + at((point) => point.avg_lag_ms), + ]} + empty="Nothing has run yet." + /> +
+
+ +
+

Flows

+
+ + + + + + + + + + + + + {(flows ?? []).map((row: FlowRollup) => ( + + + + + + + + + ))} + {flows?.length === 0 ? ( + + + + ) : null} + +
FlowExecutionsErrorsAvgLagTrend
+ + {row.flow || "—"} + + {row.executions} + {row.errors ? ( + {row.errors} failed + ) : ( + none + )} + {round(row.avg_ms)} ms{round(row.avg_lag_ms)} ms + +
+ No flow has run in the last day. +
+
+
+ +
+
+

Recent runs

+
+ {runs?.length ? ( + runs.map((run) => ( +
+ + {run.flow} + + + {run.source} + + + {run.status} + + + {round(run.duration_ms)} ms + + + {ago(run.started_at)} + +
+ )) + ) : ( +

+ No runs recorded yet. +

+ )} +
+
+ +
+

Failures

+
+ {failures?.length ? ( + failures.map((event) => ) + ) : ( +

+ Nothing has failed in the last day. +

+ )} +
+
+
+ + {dead?.length ? ( +
+

Given up on

+
+ {dead.map((item) => ( +
+ + {item.node} + + {item.reason} + + {ago(item.ts)} + +
+ ))} +
+
+ ) : null} + +
+

Changes

+
+ {audit?.length ? ( + audit.map((event) => ( +
+ + {event.actor} {event.detail} + {event.flow ? ( + + {event.flow} + + ) : null} + + + {ago(event.ts)} + +
+ )) + ) : ( +

+ Nothing has changed yet. +

+ )} +
+
+
+ ) +}