Add monitoring endpoint for oldest task in a given status

dispatcher/backend/docs/openapi_v1.yaml

@@ -1540,6 +1540,26 @@ paths:
             application/json:
               schema:
                 $ref: '#/components/schemas/Error'
+  /status/{monitor_name}:
+    get:
+      tags:
+      - public
+      summary: Get status of a given monitor
+      operationId: getMonitorStatus
+      description: Get status of a given monitor
+      parameters:
+      - $ref: '#/components/parameters/MonitorParameter'
+      - $ref: '#/components/parameters/StatusRequiredParameter'
+      - $ref: '#/components/parameters/ThresholdSecsParameter'
+      responses:
+        204:
+          description: Monitor Status
+        400:
+          description: Bad Request (invalid input)
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/InputError'
 components:
   securitySchemes:
     token:
@@ -1614,6 +1634,25 @@ components:
       required: false
       schema:
         $ref: '#/components/schemas/TaskStatus'
+    StatusRequiredParameter:
+      in: query
+      name: status
+      description: only those matching this status
+      required: true
+      schema:
+        $ref: '#/components/schemas/TaskStatus'
+    MonitorParameter:
+      in: path
+      required: true
+      name: monitor_name
+      schema:
+        $ref: '#/components/schemas/Monitor'
+    ThresholdSecsParameter:
+      in: query
+      required: true
+      name: threshold_secs
+      schema:
+        $ref: '#/components/schemas/Seconds'
   schemas:
     Language:
       type: object
@@ -2231,3 +2270,13 @@ components:
         - admin
         - worker
         - processor
+    Monitor:
+      type: string
+      example: oldest_task_older_than
+      enum:
+        - oldest_task_older_than
+    Seconds:
+      type: number
+      format: Int32
+      description: Number of seconds
+      example: 5

dispatcher/backend/src/main.py

@@ -15,6 +15,7 @@
     platforms,
     requested_tasks,
     schedules,
+    status,
     tags,
     tasks,
     users,
@@ -74,6 +75,7 @@ def home():
 application.register_blueprint(tags.Blueprint())
 application.register_blueprint(offliners.Blueprint())
 application.register_blueprint(platforms.Blueprint())
+application.register_blueprint(status.Blueprint())
 
 errors.register_handlers(application)
 

dispatcher/backend/src/routes/status/__init__.py

@@ -0,0 +1,10 @@
+from routes import API_PATH
+from routes.base import BaseBlueprint
+from routes.status.status import StatusMonitorRoute  # , StatusFooRoute
+
+
+class Blueprint(BaseBlueprint):
+    def __init__(self):
+        super().__init__("status", __name__, url_prefix=f"{API_PATH}/status")
+
+        self.register_route(StatusMonitorRoute())

dispatcher/backend/src/routes/status/status.py

@@ -0,0 +1,66 @@
+import datetime
+import logging
+
+import sqlalchemy as sa
+import sqlalchemy.orm as so
+from flask import request
+
+import db.models as dbm
+from db import dbsession
+from routes.base import BaseRoute
+from routes.errors import BadRequest
+
+logger = logging.getLogger(__name__)
+
+
+class StatusMonitorRoute(BaseRoute):
+    rule = "/<string:monitor_name>"
+    name = "status"
+    methods = ["GET"]
+
+    def oldest_task_older_than(self, session: so.Session):
+        request_args = request.args.to_dict()
+
+        if not request_args.get("threshold_secs"):
+            raise BadRequest("threshold_secs query parameter is mandatory")
+
+        if not request_args.get("status"):
+            raise BadRequest("status query parameter is mandatory")
+
+        threshold_secs = int(request_args.get("threshold_secs"))
+        status = request_args.get("status")
+
+        now = datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None)
+        min_date: datetime.datetime = min(
+            [now]
+            + session.execute(
+                (
+                    sa.select(
+                        dbm.Task.timestamp[status],
+                    ).filter(dbm.Task.status == status)
+                )
+            )
+            .scalars()
+            .all(),
+        )
+
+        return (
+            f"oldest_task_older_than for {status} and {threshold_secs}s: "
+            f"{'KO' if (now-min_date).total_seconds() > threshold_secs else 'OK'}"
+        )
+
+    @dbsession
+    def get(self, monitor_name: str, session: so.Session):
+        """Get Zimfarm status for a given monitor"""
+
+        handlers = {
+            "oldest_task_older_than": self.oldest_task_older_than,
+        }
+
+        if monitor_name not in handlers:
+            raise BadRequest(
+                f"Monitor '{monitor_name}' is not supported. Supported "
+                f"monitors: {','.join(handlers.keys())}"
+            )
+
+        return handlers[monitor_name](session)

dispatcher/backend/src/tests/integration/routes/conftest.py

@@ -453,10 +453,11 @@ def _make_task(
                 TaskStatus.requested,
                 TaskStatus.reserved,
                 TaskStatus.started,
+                TaskStatus.scraper_started,
                 TaskStatus.failed,
             ]
 
-        timestamp = {event: now for event in events}
+        timestamp = {event: now - datetime.timedelta(minutes=5) for event in events}
         events = [make_event(event, timestamp[event]) for event in events]
         container = {
             "command": "mwoffliner --mwUrl=https://example.com",
@@ -525,6 +526,7 @@ def tasks(make_task):
             make_task(status=TaskStatus.requested),
             make_task(status=TaskStatus.reserved),
             make_task(status=TaskStatus.started),
+            make_task(status=TaskStatus.scraper_started),
             make_task(status=TaskStatus.succeeded),
             make_task(status=TaskStatus.failed),
         ]

dispatcher/backend/src/tests/integration/routes/status/test_status.py

@@ -0,0 +1,60 @@
+import pytest
+
+
+class TestStatusGet:
+    url = "/status/"
+
+    @pytest.mark.parametrize(
+        "query,expected_error_part",
+        [
+            pytest.param(
+                "unknown_monitor?status=scraper_started&threshold_secs=5",
+                "Monitor 'unknown_monitor' is not supported",
+                id="unknown_monitor",
+            ),
+            pytest.param(
+                "oldest_task_older_than?status=scraper_started",
+                "threshold_secs query parameter is mandatory",
+                id="threshold_secs_missing",
+            ),
+            pytest.param(
+                "oldest_task_older_than?threshold_secs=5",
+                "status query parameter is mandatory",
+                id="status_missing",
+            ),
+        ],
+    )
+    def test_status_bad_queries(self, client, query, expected_error_part):
+        headers = {"Content-Type": "application/json"}
+        response = client.get(
+            self.url + query,
+            headers=headers,
+        )
+        assert response.status_code == 400
+        response = response.json
+        assert "error" in response
+        assert expected_error_part in response["error"]
+
+    @pytest.mark.parametrize(
+        "query,expected_reponse",
+        [
+            pytest.param(
+                "oldest_task_older_than?status=scraper_started&threshold_secs=500",
+                "oldest_task_older_than for scraper_started and 500s: OK",
+                id="oldest_task_older_than_ok",
+            ),
+            pytest.param(
+                "oldest_task_older_than?status=scraper_started&threshold_secs=5",
+                "oldest_task_older_than for scraper_started and 5s: KO",
+                id="oldest_task_older_than_ko",
+            ),
+        ],
+    )
+    def test_status_normal_queries(self, client, tasks, query, expected_reponse):
+        headers = {"Content-Type": "application/json"}
+        response = client.get(
+            self.url + "oldest_task_older_than?status=scraper_started&threshold_secs=5",
+            headers=headers,
+        )
+        assert response.status_code == 200
+        assert response.text == "oldest_task_older_than for scraper_started and 5s: KO"