From 541e2bfe6a9877b0e49cdc3b5ab7669f7cba7512 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Sun, 24 May 2026 21:04:34 -0400 Subject: [PATCH] =?UTF-8?q?fc3i:=20/api/system/activity=20blueprint=20?= =?UTF-8?q?=E2=80=94=20queues,=20workers,=20runs,=20failures?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 4.7 (1M context) --- backend/app/api/__init__.py | 2 + backend/app/api/system_activity.py | 209 +++++++++++++++++++++++++++++ 2 files changed, 211 insertions(+) create mode 100644 backend/app/api/system_activity.py diff --git a/backend/app/api/__init__.py b/backend/app/api/__init__.py index 62ad522..441d58d 100644 --- a/backend/app/api/__init__.py +++ b/backend/app/api/__init__.py @@ -33,6 +33,7 @@ def all_blueprints() -> list[Blueprint]: from .showcase import showcase_bp from .sources import sources_bp from .suggestions import suggestions_bp + from .system_activity import system_activity_bp from .tags import tags_bp return [ api_bp, @@ -44,6 +45,7 @@ def all_blueprints() -> list[Blueprint]: artists_bp, showcase_bp, settings_bp, + system_activity_bp, import_admin_bp, migrate_bp, suggestions_bp, diff --git a/backend/app/api/system_activity.py b/backend/app/api/system_activity.py new file mode 100644 index 0000000..480b8c9 --- /dev/null +++ b/backend/app/api/system_activity.py @@ -0,0 +1,209 @@ +"""FC-3i: system activity dashboard endpoints. + +Read-only. Combines Redis-broker queue depths (LLEN per queue), +Celery worker introspection (celery inspect), and the task_run DB +history into the surfaces the SystemActivityTab UI consumes. + +All filesystem/sync-client work goes through asyncio.to_thread per +ASYNC230/240 (mirrors backend.app.api.extension's pattern). +""" + +from __future__ import annotations + +import asyncio +import time +from datetime import UTC, datetime, timedelta + +from quart import Blueprint, jsonify, request +from sqlalchemy import desc, func, select + +from ..config import get_config +from ..extensions import get_session +from ..models import TaskRun + +system_activity_bp = Blueprint( + "system_activity", __name__, url_prefix="/api/system/activity", +) + +# Canonical queue order — must match celery_app.task_routes. UI renders +# in this order; queues with no LLEN response show as null rather than +# absent. +_QUEUE_NAMES = ( + "default", "import", "thumbnail", "ml", + "download", "scan", "maintenance", +) + +# Cache module-level so all requests share the cache between polls. +# Tests can reset via direct dict mutation if needed. +_QUEUE_CACHE: dict = {"ts": 0.0, "data": None} +_WORKER_CACHE: dict = {"ts": 0.0, "data": None} +_QUEUE_CACHE_TTL = 2.0 +_WORKER_CACHE_TTL = 5.0 + + +def _read_queues_sync() -> dict: + """Reads each queue's LLEN from the broker. Sync — caller wraps in + asyncio.to_thread. Per-queue try/except returns None on failure so + one bad queue doesn't break the whole response.""" + import redis # local import; only this endpoint needs it + + cfg = get_config() + client = redis.Redis.from_url(cfg.celery_broker_url) + out: dict = {} + for name in _QUEUE_NAMES: + try: + out[name] = int(client.llen(name)) + except Exception: # noqa: BLE001 — broker hiccup shouldn't break UI + out[name] = None + return { + "queues": out, + "fetched_at": datetime.now(UTC).isoformat(), + } + + +def _read_workers_sync() -> dict: + """celery inspect active_queues + active. Returns per-worker info.""" + from ..celery_app import celery as celery_app + + insp = celery_app.control.inspect(timeout=2.0) + active_queues = insp.active_queues() or {} + active_tasks = insp.active() or {} + + workers: dict = {} + for hostname, queues in active_queues.items(): + workers[hostname] = { + "queues": sorted({q["name"] for q in queues}), + "active_count": len(active_tasks.get(hostname, [])), + } + return { + "workers": workers, + "fetched_at": datetime.now(UTC).isoformat(), + } + + +@system_activity_bp.route("/queues", methods=["GET"]) +async def get_queues(): + """Per-queue Redis LLEN. Cached 2s. + + Response: {queues: {name: depth_or_null}, fetched_at: iso8601} + """ + now = time.time() + if _QUEUE_CACHE["data"] is None or (now - _QUEUE_CACHE["ts"]) > _QUEUE_CACHE_TTL: + _QUEUE_CACHE["data"] = await asyncio.to_thread(_read_queues_sync) + _QUEUE_CACHE["ts"] = now + return jsonify(_QUEUE_CACHE["data"]) + + +@system_activity_bp.route("/workers", methods=["GET"]) +async def get_workers(): + """Live celery inspect. Cached 5s. + + Response: {workers: {hostname: {queues, active_count}}, fetched_at} + """ + now = time.time() + if _WORKER_CACHE["data"] is None or (now - _WORKER_CACHE["ts"]) > _WORKER_CACHE_TTL: + _WORKER_CACHE["data"] = await asyncio.to_thread(_read_workers_sync) + _WORKER_CACHE["ts"] = now + return jsonify(_WORKER_CACHE["data"]) + + +@system_activity_bp.route("/runs", methods=["GET"]) +async def list_runs(): + """Paginated task_run history. Query params: + queue= filter to one queue + status= filter to one status (running/ok/error/timeout/retry) + limit= default 50, max 200 + before_id= cursor for keyset pagination + + Response: {runs: [...], next_cursor: id|null} + """ + try: + limit = min(int(request.args.get("limit", "50")), 200) + except ValueError: + return jsonify({"error": "invalid_limit"}), 400 + if limit < 1: + return jsonify({"error": "invalid_limit"}), 400 + + queue = request.args.get("queue") + status = request.args.get("status") + before_id_raw = request.args.get("before_id") + before_id = int(before_id_raw) if before_id_raw else None + + async with get_session() as session: + stmt = select(TaskRun).order_by(desc(TaskRun.id)) + if queue: + stmt = stmt.where(TaskRun.queue == queue) + if status: + stmt = stmt.where(TaskRun.status == status) + if before_id is not None: + stmt = stmt.where(TaskRun.id < before_id) + stmt = stmt.limit(limit + 1) + rows = (await session.execute(stmt)).scalars().all() + + has_more = len(rows) > limit + rows = rows[:limit] + return jsonify({ + "runs": [_row_to_dict(r) for r in rows], + "next_cursor": rows[-1].id if has_more and rows else None, + }) + + +@system_activity_bp.route("/failures", methods=["GET"]) +async def list_failures(): + """Recent failures across all lanes (24h window). + + Response: {recent: [...], count_by_type: {ErrorClass: n}, since} + """ + try: + limit = min(int(request.args.get("limit", "50")), 200) + except ValueError: + return jsonify({"error": "invalid_limit"}), 400 + + since = datetime.now(UTC) - timedelta(hours=24) + + async with get_session() as session: + recent_stmt = ( + select(TaskRun) + .where(TaskRun.status.in_(["error", "timeout"])) + .where(TaskRun.finished_at >= since) + .order_by(desc(TaskRun.finished_at)) + .limit(limit) + ) + recent = (await session.execute(recent_stmt)).scalars().all() + + count_stmt = ( + select(TaskRun.error_type, func.count(TaskRun.id)) + .where(TaskRun.status.in_(["error", "timeout"])) + .where(TaskRun.finished_at >= since) + .group_by(TaskRun.error_type) + .order_by(desc(func.count(TaskRun.id))) + ) + counts = (await session.execute(count_stmt)).all() + + return jsonify({ + "recent": [_row_to_dict(r) for r in recent], + "count_by_type": { + (row[0] or "Unknown"): row[1] + for row in counts + }, + "since": since.isoformat(), + }) + + +def _row_to_dict(r: TaskRun) -> dict: + return { + "id": r.id, + "queue": r.queue, + "task_name": r.task_name, + "target_id": r.target_id, + "celery_task_id": r.celery_task_id, + "started_at": r.started_at.isoformat() if r.started_at else None, + "finished_at": r.finished_at.isoformat() if r.finished_at else None, + "duration_ms": r.duration_ms, + "status": r.status, + "error_type": r.error_type, + "error_message": r.error_message, + "retry_count": r.retry_count, + "worker_hostname": r.worker_hostname, + "args_summary": r.args_summary, + }