diff --git a/app/api/__init__.py b/app/api/__init__.py index b483248a..97f13099 100644 --- a/app/api/__init__.py +++ b/app/api/__init__.py @@ -15,6 +15,7 @@ from app.api.logs import router as logs_router from app.api.onedrive import router as onedrive_router from app.api.openai import router as openai_router from app.api.process import router as process_router +from app.api.queue import router as queue_router from app.api.search import router as search_router from app.api.settings import router as settings_router from app.api.url_upload import router as url_upload_router @@ -42,3 +43,4 @@ router.include_router(logs_router) router.include_router(settings_router) router.include_router(url_upload_router) router.include_router(search_router) +router.include_router(queue_router) diff --git a/app/api/queue.py b/app/api/queue.py new file mode 100644 index 00000000..fdc851af --- /dev/null +++ b/app/api/queue.py @@ -0,0 +1,270 @@ +""" +Queue monitoring API endpoints. + +Provides endpoints to query Celery/Redis queue statistics and +database-level processing status for document pipeline visibility. +""" + +import logging +from typing import Any + +import redis +from fastapi import APIRouter, Depends +from sqlalchemy import func +from sqlalchemy.orm import Session + +from app.config import settings +from app.database import get_db +from app.models import FileProcessingStep, FileRecord + +logger = logging.getLogger(__name__) +router = APIRouter(prefix="/queue", tags=["queue"]) + +# Constants +CELERY_INSPECT_TIMEOUT = 2.0 +MAX_ARGS_DISPLAY_LENGTH = 200 + + +def _get_redis_queue_length(redis_client: redis.Redis, queue_name: str) -> int: + """Get the number of messages in a Redis-backed Celery queue. + + Args: + redis_client: Connected Redis client instance. + queue_name: Name of the Celery queue to inspect. + + Returns: + Number of messages (tasks) waiting in the queue. + """ + try: + return redis_client.llen(queue_name) + except Exception: + logger.debug(f"Could not read queue length for '{queue_name}'") + return 0 + + +def _get_celery_inspect_stats() -> dict[str, Any]: + """Query the Celery inspect API for active, reserved, and scheduled tasks. + + Returns: + Dictionary with active, reserved, and scheduled task summaries. + """ + from app.celery_app import celery + + result: dict[str, Any] = { + "active": [], + "reserved": [], + "scheduled": [], + "workers_online": 0, + } + + try: + inspector = celery.control.inspect(timeout=CELERY_INSPECT_TIMEOUT) + + active = inspector.active() or {} + reserved = inspector.reserved() or {} + scheduled = inspector.scheduled() or {} + + result["workers_online"] = len(active) + + for _worker, tasks in active.items(): + for task in tasks: + result["active"].append( + { + "id": task.get("id", ""), + "name": task.get("name", "unknown"), + "args": str(task.get("args", []))[:MAX_ARGS_DISPLAY_LENGTH], + "started": task.get("time_start"), + } + ) + + for _worker, tasks in reserved.items(): + for task in tasks: + result["reserved"].append( + { + "id": task.get("id", ""), + "name": task.get("name", "unknown"), + "args": str(task.get("args", []))[:MAX_ARGS_DISPLAY_LENGTH], + } + ) + + for _worker, tasks in scheduled.items(): + for task in tasks: + req = task.get("request", {}) + result["scheduled"].append( + { + "id": req.get("id", ""), + "name": req.get("name", "unknown"), + "eta": task.get("eta"), + } + ) + except Exception as exc: + logger.warning(f"Celery inspect failed (workers may be offline): {exc}") + + return result + + +def _get_db_processing_summary(db: Session) -> dict[str, Any]: + """Query the database for a summary of file processing states. + + Args: + db: SQLAlchemy database session. + + Returns: + Dictionary with counts of files by processing state. + """ + try: + total_files = db.query(func.count(FileRecord.id)).scalar() or 0 + + # Count files with at least one in_progress step + processing_count = ( + db.query(func.count(func.distinct(FileProcessingStep.file_id))) + .filter(FileProcessingStep.status == "in_progress") + .scalar() + or 0 + ) + + # Count files with at least one failure and no in_progress + failed_subq = ( + db.query(FileProcessingStep.file_id).filter(FileProcessingStep.status == "failure").distinct().subquery() + ) + in_progress_subq = ( + db.query(FileProcessingStep.file_id) + .filter(FileProcessingStep.status == "in_progress") + .distinct() + .subquery() + ) + failed_count = ( + db.query(func.count(func.distinct(failed_subq.c.file_id))) + .filter(~failed_subq.c.file_id.in_(db.query(in_progress_subq.c.file_id))) + .scalar() + or 0 + ) + + # Count files that have steps and all steps are success/skipped + all_step_files = db.query(FileProcessingStep.file_id).distinct().subquery() + # Files with any non-terminal step + non_terminal = ( + db.query(FileProcessingStep.file_id) + .filter(FileProcessingStep.status.in_(["in_progress", "pending", "failure"])) + .distinct() + .subquery() + ) + completed_count = ( + db.query(func.count(func.distinct(all_step_files.c.file_id))) + .filter(~all_step_files.c.file_id.in_(db.query(non_terminal.c.file_id))) + .scalar() + or 0 + ) + + # Files with no processing steps at all + files_with_steps = db.query(FileProcessingStep.file_id).distinct().subquery() + pending_count = ( + db.query(func.count(FileRecord.id)) + .filter(~FileRecord.id.in_(db.query(files_with_steps.c.file_id))) + .filter(FileRecord.is_duplicate.is_(False)) + .scalar() + or 0 + ) + + # Recent files being processed (last 20 in_progress or pending) + recent_processing = ( + db.query(FileRecord.id, FileRecord.original_filename, FileProcessingStep.step_name) + .join(FileProcessingStep, FileRecord.id == FileProcessingStep.file_id) + .filter(FileProcessingStep.status == "in_progress") + .order_by(FileProcessingStep.updated_at.desc()) + .limit(20) + .all() + ) + + recent_list = [ + {"file_id": r[0], "filename": r[1] or f"File #{r[0]}", "current_step": r[2]} for r in recent_processing + ] + + return { + "total_files": total_files, + "processing": processing_count, + "failed": failed_count, + "completed": completed_count, + "pending": pending_count, + "recent_processing": recent_list, + } + except Exception as exc: + logger.error(f"Error querying DB processing summary: {exc}") + return { + "total_files": 0, + "processing": 0, + "failed": 0, + "completed": 0, + "pending": 0, + "recent_processing": [], + } + + +@router.get("/stats") +def get_queue_stats(db: Session = Depends(get_db)) -> dict[str, Any]: + """Get comprehensive queue and processing statistics. + + Returns queue lengths from Redis, Celery worker inspection data, + and database-level processing summaries for the document pipeline. + + Returns: + Dictionary containing redis queue info, celery worker info, + and database processing summary. + """ + # 1. Redis queue lengths + queue_lengths: dict[str, int] = {} + try: + redis_client = redis.Redis.from_url(settings.redis_url, decode_responses=True) + for queue_name in ["document_processor", "default", "celery"]: + queue_lengths[queue_name] = _get_redis_queue_length(redis_client, queue_name) + redis_client.close() + except Exception as exc: + logger.warning(f"Could not connect to Redis: {exc}") + + total_queued = sum(queue_lengths.values()) + + # 2. Celery inspect + celery_stats = _get_celery_inspect_stats() + + # 3. DB summary + db_summary = _get_db_processing_summary(db) + + return { + "queues": queue_lengths, + "total_queued": total_queued, + "celery": celery_stats, + "db_summary": db_summary, + } + + +@router.get("/pending-count") +def get_pending_count(db: Session = Depends(get_db)) -> dict[str, int]: + """Get a lightweight count of queued + in-progress items for the files page banner. + + Returns: + Dictionary with total_pending count (queued in Redis + processing in DB). + """ + total_pending = 0 + + # Redis queue lengths + try: + redis_client = redis.Redis.from_url(settings.redis_url, decode_responses=True) + for queue_name in ["document_processor", "default", "celery"]: + total_pending += _get_redis_queue_length(redis_client, queue_name) + redis_client.close() + except Exception: + logger.debug("Could not connect to Redis for pending count") + + # DB in-progress count + try: + processing_count = ( + db.query(func.count(func.distinct(FileProcessingStep.file_id))) + .filter(FileProcessingStep.status == "in_progress") + .scalar() + or 0 + ) + total_pending += processing_count + except Exception: + logger.debug("Could not query DB for processing count") + + return {"total_pending": total_pending} diff --git a/app/views/__init__.py b/app/views/__init__.py index bbc00bab..ed55ff29 100644 --- a/app/views/__init__.py +++ b/app/views/__init__.py @@ -12,6 +12,7 @@ from app.views.general import router as general_router from app.views.google_drive import router as google_drive_router from app.views.license_routes import router as license_router # Add the license router from app.views.onedrive import router as onedrive_router +from app.views.queue import router as queue_router from app.views.search import router as search_router from app.views.settings import router as settings_router from app.views.status import router as status_router @@ -29,3 +30,4 @@ router.include_router(license_router) # Include the license router router.include_router(settings_router) router.include_router(filemanager_router) router.include_router(search_router) +router.include_router(queue_router) diff --git a/app/views/queue.py b/app/views/queue.py new file mode 100644 index 00000000..993fde4f --- /dev/null +++ b/app/views/queue.py @@ -0,0 +1,32 @@ +""" +Queue monitoring view for the admin dashboard. +""" + +import logging + +from fastapi import Depends, Request +from sqlalchemy.orm import Session + +from app.views.base import APIRouter, get_db, require_login, templates +from app.views.settings import require_admin_access + +logger = logging.getLogger(__name__) +router = APIRouter() + + +@router.get("/admin/queue") +@require_login +@require_admin_access +async def queue_dashboard(request: Request, db: Session = Depends(get_db)): + """ + Queue monitoring dashboard — admin only. + + Displays Celery/Redis queue statistics and database processing summaries + so administrators can monitor the document processing pipeline. + """ + return templates.TemplateResponse( + "queue_dashboard.html", + { + "request": request, + }, + ) diff --git a/docs/API.md b/docs/API.md index 4255f02b..f0d6f87e 100644 --- a/docs/API.md +++ b/docs/API.md @@ -437,6 +437,57 @@ Errors follow standard HTTP status codes with descriptive messages: } ``` +## Queue Monitoring + +### GET /api/queue/stats + +Get comprehensive queue and processing statistics, including Redis queue lengths, Celery worker inspection data, and database-level processing summaries. + +**Authentication:** Required + +**Response (200 OK):** +```json +{ + "queues": { + "document_processor": 12, + "default": 0, + "celery": 0 + }, + "total_queued": 12, + "celery": { + "active": [ + {"id": "abc123", "name": "process_document", "args": "[42]", "started": 1700000000} + ], + "reserved": [], + "scheduled": [], + "workers_online": 1 + }, + "db_summary": { + "total_files": 5000, + "processing": 3, + "failed": 1, + "completed": 4900, + "pending": 96, + "recent_processing": [ + {"file_id": 42, "filename": "invoice.pdf", "current_step": "extract_metadata_with_gpt"} + ] + } +} +``` + +### GET /api/queue/pending-count + +Lightweight endpoint returning the total number of queued + in-progress items. Designed for the files page banner indicator. + +**Authentication:** Required + +**Response (200 OK):** +```json +{ + "total_pending": 15 +} +``` + ## Rate Limiting The API implements rate limiting to ensure system stability. If you exceed the limits, you'll receive a `429 Too Many Requests` response. diff --git a/docs/UserGuide.md b/docs/UserGuide.md index 8f1d5921..39c83cb4 100644 --- a/docs/UserGuide.md +++ b/docs/UserGuide.md @@ -92,6 +92,23 @@ The **Files** page provides access to all processed documents: 3. Click on any file to view its details 4. Sort the list by any column by clicking on the column header +> **Tip:** When documents are being processed, a blue banner appears at the top of the Files page showing how many items are queued or currently processing. Files will appear in the list once their processing completes. Admins can click "View Queue" in the banner to open the Queue Monitor dashboard. + +## Queue Monitor (Admin) + +The **Queue Monitor** dashboard provides real-time visibility into the document processing pipeline. It is available to admin users under **Admin → Queue Monitor** in the navigation bar. + +The dashboard shows: +- **Queued Tasks** — number of tasks waiting in Redis-backed Celery queues +- **Active Tasks** — tasks currently being executed by Celery workers +- **Files Processing** — files with at least one in-progress processing step +- **Workers Online** — number of connected Celery worker processes +- **Redis Queues** — per-queue breakdown of pending task counts +- **Processing Pipeline** — database-level summary of file states (completed, processing, pending, failed) +- **Recently Processing Files** — the most recent files being actively processed, with links to their detail pages + +The dashboard auto-refreshes every 10 seconds. + ## Searching Documents DocuElevate provides two ways to search your documents: diff --git a/frontend/templates/base.html b/frontend/templates/base.html index ebe09847..75dcb274 100644 --- a/frontend/templates/base.html +++ b/frontend/templates/base.html @@ -95,6 +95,9 @@ File Manager + + Queue Monitor + @@ -168,6 +171,9 @@ File Manager + + Queue Monitor + diff --git a/frontend/templates/files.html b/frontend/templates/files.html index cc4e62ae..9c9b13aa 100644 --- a/frontend/templates/files.html +++ b/frontend/templates/files.html @@ -370,6 +370,23 @@

File Records

+ + + {% if error %}

Error: {{ error }}

@@ -1050,5 +1067,35 @@ _searchCurrentPage = 1; } + + +
{% endblock %} diff --git a/frontend/templates/queue_dashboard.html b/frontend/templates/queue_dashboard.html new file mode 100644 index 00000000..65c9fdea --- /dev/null +++ b/frontend/templates/queue_dashboard.html @@ -0,0 +1,320 @@ +{% extends "base.html" %} +{% block title %}Queue Monitor — DocuElevate{% endblock %} + +{% block head_extra %} + +{% endblock %} + +{% block content %} +
+ + +
+

+ Queue Monitor +

+

Real-time view of the document processing pipeline and Celery task queues.

+
+ + +
+ +

Loading queue statistics…

+
+ + + + + + +
+ + +{% endblock %} diff --git a/tests/test_queue_monitoring.py b/tests/test_queue_monitoring.py new file mode 100644 index 00000000..9f9f5ad7 --- /dev/null +++ b/tests/test_queue_monitoring.py @@ -0,0 +1,246 @@ +"""Tests for app/api/queue.py and app/views/queue.py modules.""" + +from unittest.mock import MagicMock, Mock, patch + +import pytest + + +@pytest.mark.unit +class TestGetRedisQueueLength: + """Tests for the _get_redis_queue_length helper.""" + + def test_returns_queue_length(self): + """Test returns queue length from Redis.""" + from app.api.queue import _get_redis_queue_length + + mock_redis = MagicMock() + mock_redis.llen.return_value = 42 + assert _get_redis_queue_length(mock_redis, "document_processor") == 42 + mock_redis.llen.assert_called_once_with("document_processor") + + def test_returns_zero_on_error(self): + """Test returns 0 when Redis call fails.""" + from app.api.queue import _get_redis_queue_length + + mock_redis = MagicMock() + mock_redis.llen.side_effect = Exception("Connection refused") + assert _get_redis_queue_length(mock_redis, "default") == 0 + + +@pytest.mark.unit +class TestGetCeleryInspectStats: + """Tests for the _get_celery_inspect_stats helper.""" + + @patch("app.celery_app.celery") + def test_returns_worker_stats(self, mock_celery_mod): + """Test returns active, reserved, scheduled tasks.""" + from app.api.queue import _get_celery_inspect_stats + + mock_inspector = MagicMock() + mock_inspector.active.return_value = { + "worker1": [ + {"id": "task-1", "name": "app.tasks.process_document.process_document", "args": [1], "time_start": 123} + ] + } + mock_inspector.reserved.return_value = { + "worker1": [{"id": "task-2", "name": "app.tasks.upload_to_s3.upload_to_s3", "args": [2]}] + } + mock_inspector.scheduled.return_value = { + "worker1": [ + { + "request": {"id": "task-3", "name": "app.tasks.check_credentials.check_credentials"}, + "eta": "2026-01-01", + } + ] + } + mock_celery_mod.control.inspect.return_value = mock_inspector + + result = _get_celery_inspect_stats() + + assert result["workers_online"] == 1 + assert len(result["active"]) == 1 + assert result["active"][0]["id"] == "task-1" + assert len(result["reserved"]) == 1 + assert len(result["scheduled"]) == 1 + + @patch("app.celery_app.celery") + def test_handles_no_workers(self, mock_celery_mod): + """Test handles case where no workers are online.""" + from app.api.queue import _get_celery_inspect_stats + + mock_inspector = MagicMock() + mock_inspector.active.return_value = None + mock_inspector.reserved.return_value = None + mock_inspector.scheduled.return_value = None + mock_celery_mod.control.inspect.return_value = mock_inspector + + result = _get_celery_inspect_stats() + + assert result["workers_online"] == 0 + assert result["active"] == [] + assert result["reserved"] == [] + assert result["scheduled"] == [] + + @patch("app.celery_app.celery") + def test_handles_inspect_exception(self, mock_celery_mod): + """Test handles exception during inspect.""" + from app.api.queue import _get_celery_inspect_stats + + mock_celery_mod.control.inspect.side_effect = Exception("Broker unreachable") + + result = _get_celery_inspect_stats() + + assert result["workers_online"] == 0 + assert result["active"] == [] + + +@pytest.mark.unit +class TestGetDbProcessingSummary: + """Tests for the _get_db_processing_summary helper.""" + + def test_returns_summary(self, db_session): + """Test returns processing summary from DB.""" + from app.api.queue import _get_db_processing_summary + from app.models import FileProcessingStep, FileRecord + + # Create some test files + file1 = FileRecord(filehash="abc1", local_filename="f1.pdf", file_size=100, is_duplicate=False) + file2 = FileRecord(filehash="abc2", local_filename="f2.pdf", file_size=200, is_duplicate=False) + db_session.add_all([file1, file2]) + db_session.commit() + + # Add steps: file1 completed, file2 in_progress + step1 = FileProcessingStep(file_id=file1.id, step_name="extract_text", status="success") + step2 = FileProcessingStep(file_id=file2.id, step_name="extract_text", status="in_progress") + db_session.add_all([step1, step2]) + db_session.commit() + + result = _get_db_processing_summary(db_session) + + assert result["total_files"] == 2 + assert result["processing"] == 1 + assert isinstance(result["recent_processing"], list) + + def test_returns_empty_on_error(self): + """Test returns empty summary on DB error.""" + from app.api.queue import _get_db_processing_summary + + mock_db = MagicMock() + mock_db.query.side_effect = Exception("DB error") + + result = _get_db_processing_summary(mock_db) + + assert result["total_files"] == 0 + assert result["processing"] == 0 + assert result["recent_processing"] == [] + + +@pytest.mark.integration +class TestQueueStatsEndpoint: + """Tests for the GET /api/queue/stats endpoint.""" + + @patch("app.api.queue.redis.Redis") + @patch("app.celery_app.celery") + def test_queue_stats_returns_200(self, mock_celery_mod, mock_redis_cls, client): + """Test queue stats endpoint returns 200 with data.""" + # Mock Redis + mock_redis_instance = MagicMock() + mock_redis_instance.llen.return_value = 5 + mock_redis_cls.from_url.return_value = mock_redis_instance + + # Mock Celery inspector + mock_inspector = MagicMock() + mock_inspector.active.return_value = {} + mock_inspector.reserved.return_value = {} + mock_inspector.scheduled.return_value = {} + mock_celery_mod.control.inspect.return_value = mock_inspector + + response = client.get("/api/queue/stats") + assert response.status_code == 200 + + data = response.json() + assert "queues" in data + assert "total_queued" in data + assert "celery" in data + assert "db_summary" in data + + @patch("app.api.queue.redis.Redis") + def test_queue_stats_handles_redis_error(self, mock_redis_cls, client): + """Test queue stats handles Redis connection failure.""" + mock_redis_cls.from_url.side_effect = Exception("Redis unavailable") + + response = client.get("/api/queue/stats") + assert response.status_code == 200 + data = response.json() + assert data["total_queued"] == 0 + + +@pytest.mark.integration +class TestPendingCountEndpoint: + """Tests for the GET /api/queue/pending-count endpoint.""" + + @patch("app.api.queue.redis.Redis") + def test_pending_count_returns_200(self, mock_redis_cls, client): + """Test pending count endpoint returns 200.""" + mock_redis_instance = MagicMock() + mock_redis_instance.llen.return_value = 3 + mock_redis_cls.from_url.return_value = mock_redis_instance + + response = client.get("/api/queue/pending-count") + assert response.status_code == 200 + + data = response.json() + assert "total_pending" in data + assert data["total_pending"] >= 0 + + @patch("app.api.queue.redis.Redis") + def test_pending_count_redis_failure_still_works(self, mock_redis_cls, client): + """Test pending count still works when Redis is down.""" + mock_redis_cls.from_url.side_effect = Exception("Connection refused") + + response = client.get("/api/queue/pending-count") + assert response.status_code == 200 + data = response.json() + assert "total_pending" in data + + +@pytest.mark.integration +class TestQueueDashboardView: + """Tests for the GET /admin/queue view.""" + + def test_queue_dashboard_requires_login(self, client): + """Test queue dashboard requires authentication.""" + response = client.get("/admin/queue", follow_redirects=False) + assert response.status_code in [200, 302, 401] + + def test_queue_dashboard_non_admin_redirect(self, client): + """Test queue dashboard redirects non-admin users.""" + # Set non-admin session + with client: + client.cookies.set("session", "test") + response = client.get("/admin/queue", follow_redirects=False) + # Should redirect or deny since no admin session + assert response.status_code in [200, 302, 401] + + +@pytest.mark.unit +class TestQueueDashboardViewFunction: + """Tests for the queue_dashboard view function.""" + + @patch("app.views.queue.templates") + @pytest.mark.asyncio + async def test_queue_dashboard_returns_template(self, mock_templates): + """Test queue dashboard returns template response.""" + from app.views.queue import queue_dashboard + + mock_request = Mock() + mock_request.session = {"user": {"is_admin": True}} + mock_db = MagicMock() + + result = await queue_dashboard(mock_request, db=mock_db) + + mock_templates.TemplateResponse.assert_called_once() + call_args = mock_templates.TemplateResponse.call_args + assert call_args[0][0] == "queue_dashboard.html" + context = call_args[0][1] + assert "request" in context