Update/dockerflow 2024.04.2, separate queue checks (#974)

* Updating to dockerflow 2024.04.2
* Making queue checks async, separating queue checks
Co-authored-by: Mathieu Leplatre <[email protected]>

Author: alexcottner

jbi/app.py

@@ -91,7 +91,8 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
         name="jira.all_project_issue_types_exist",
     )
     checks.register(jira_service.check_jira_pandoc_install, name="jira.pandoc_install")
-    checks.register(queue.ready, name="queue.ready")
+    checks.register(queue.check_writable, name="queue.writable")
+    checks.register(queue.check_readable, name="queue.readable")
 
     yield
 

jbi/queue.py

@@ -22,16 +22,16 @@
       item and parse it as an item
 """
 
-import asyncio
 import logging
+import re
 import tempfile
 import traceback
 from abc import ABC, abstractmethod
 from datetime import datetime
 from functools import lru_cache
 from json import JSONDecodeError
 from pathlib import Path
-from typing import Any, AsyncIterator, Optional
+from typing import AsyncIterator, Optional
 from urllib.parse import ParseResult, urlparse
 
 import dockerflow.checks
@@ -43,12 +43,13 @@
 logger = logging.getLogger(__name__)
 
 
-async def async_iter(iter: AsyncIterator[Any]) -> list[Any]:
-    return [item async for item in iter]
-
-
 class QueueItemRetrievalError(Exception):
-    pass
+    def __init__(self, message=None, path=None):
+        self.message = message or "Error reading or parsing queue item"
+        self.path = path
+
+    def __str__(self):
+        return f"QueueItemRetrievalError: {self.message} - path: {self.path}."
 
 
 class InvalidQueueDSNError(Exception):
@@ -199,13 +200,15 @@ async def get(self, bug_id: int) -> AsyncIterator[QueueItem]:
                 yield QueueItem.parse_file(path)
             except (JSONDecodeError, ValidationError) as e:
                 raise QueueItemRetrievalError(
-                    f"Unable to load item at path {path} from queue"
+                    "Unable to load item from queue", path=path
                 ) from e
 
     async def get_all(self) -> dict[int, AsyncIterator[QueueItem]]:
         all_items: dict[int, AsyncIterator[QueueItem]] = {}
         for filesystem_object in self.location.iterdir():
-            if filesystem_object.is_dir():
+            if filesystem_object.is_dir() and re.match(
+                "\d", filesystem_object.name
+            ):  # filtering out temp files from checks
                 all_items[int(filesystem_object.name)] = self.get(filesystem_object)
         return all_items
 
@@ -224,12 +227,8 @@ def __init__(self, dsn: FileUrl | str | ParseResult):
             raise InvalidQueueDSNError(f"{dsn.scheme} is not supported")
         self.backend = FileBackend(dsn.path)
 
-    def ready(self) -> list[dockerflow.checks.CheckMessage]:
-        """Heartbeat check to assert we can write items to queue
-
-        TODO: Convert to an async method when Dockerflow's FastAPI integration
-        can run check asynchronously
-        """
+    def check_writable(self) -> list[dockerflow.checks.CheckMessage]:
+        """Heartbeat check to assert we can write items to queue"""
         results = []
         ping_result = self.backend.ping()
         if ping_result is False:
@@ -240,11 +239,25 @@ def ready(self) -> list[dockerflow.checks.CheckMessage]:
                     id="queue.backend.ping",
                 )
             )
+        return results
 
+    async def check_readable(self) -> list[dockerflow.checks.CheckMessage]:
+        results = []
         try:
-            bugs_items = asyncio.run(self.retrieve())
-            for items in bugs_items.values():
-                asyncio.run(async_iter(items))
+            bugs = await self.retrieve()
+
+            for bug_id, items in bugs.items():
+                try:
+                    bug_items = (await self.retrieve()).values()
+                    [[i async for i in items] for items in bug_items]
+                except QueueItemRetrievalError as exc:
+                    results.append(
+                        dockerflow.checks.Error(
+                            f"failed to parse file {str(exc.path)}",
+                            hint="check that parked event files are not corrupt",
+                            id="queue.backend.read",
+                        )
+                    )
         except Exception as exc:
             logger.exception(exc)
             results.append(
@@ -254,7 +267,6 @@ def ready(self) -> list[dockerflow.checks.CheckMessage]:
                     id="queue.backend.retrieve",
                 )
             )
-
         return results
 
     async def postpone(self, payload: bugzilla.WebhookRequest) -> None:

poetry.lock

@@ -11,7 +11,7 @@ fastapi = "^0.110.2"
 pydantic = {version = "^2.7.0", extras = ["email"]}
 uvicorn = {extras = ["standard"], version = "^0.29.0"}
 atlassian-python-api = "^3.41.11"
-dockerflow = {extras = ["fastapi"], version = "2024.4.1"}
+dockerflow = {extras = ["fastapi"], version = "2024.4.2"}
 Jinja2 = "^3.1.3"
 sentry-sdk = {extras = ["fastapi"], version = "^1.45.0"}
 pydantic-yaml = "^1.3.0"

pyproject.toml

@@ -149,6 +149,20 @@ async def test_backend_get_all_invalid_json(backend: QueueBackend, queue_item_fa
     assert len(items) == 2
 
 
+@pytest.mark.asyncio
+async def test_backend_get_all_ignores_bad_folders(
+    backend: QueueBackend, queue_item_factory
+):
+    item_1 = queue_item_factory()
+    await backend.put(item_1)
+
+    corrupt_file_dir = backend.location / "abc"
+    corrupt_file_dir.mkdir()
+
+    items = await backend.get_all()
+    assert len(items) == 1
+
+
 @pytest.mark.asyncio
 async def test_backend_get_all_payload_doesnt_match_schema(
     backend: QueueBackend, queue_item_factory
@@ -179,6 +193,7 @@ async def test_backend_get_invalid_json(backend: QueueBackend, queue_item_factor
         await anext(items)
 
 
+@pytest.mark.asyncio
 async def test_get_missing_timezone(backend: QueueBackend, queue_item_factory):
     item = queue_item_factory.build(payload__bug__id=666)
     dump = item.model_dump()
@@ -196,6 +211,7 @@ async def test_get_missing_timezone(backend: QueueBackend, queue_item_factory):
     assert "2024-04-18T12:46:54Z" in item.model_dump_json(), "timezone put in dump"
 
 
+@pytest.mark.asyncio
 async def test_backend_get_payload_doesnt_match_schema(
     backend: QueueBackend, queue_item_factory
 ):
@@ -211,26 +227,32 @@ async def test_backend_get_payload_doesnt_match_schema(
         await anext(items)
 
 
-def test_ready_ok(queue: DeadLetterQueue):
-    assert queue.ready() == []
+def test_check_writable_ok(queue: DeadLetterQueue):
+    assert queue.check_writable() == []
 
 
-def test_ready_not_writable(queue: DeadLetterQueue, tmp_path):
+def test_check_writable_not_writable(queue: DeadLetterQueue, tmp_path):
     queue.backend = FileBackend(tmp_path)
     tmp_path.chmod(0o400)  # set to readonly
-    [failure] = queue.ready()
+    [failure] = queue.check_writable()
     assert failure.id == "queue.backend.ping"
 
 
-def test_ready_not_parseable(queue: DeadLetterQueue):
+@pytest.mark.asyncio
+async def test_check_readable_ok(queue: DeadLetterQueue):
+    assert await queue.check_readable() == []
+
+
+@pytest.mark.asyncio
+async def test_check_readable_not_parseable(queue: DeadLetterQueue):
     corrupt_file_dir = queue.backend.location / "999"
     corrupt_file_dir.mkdir()
     corrupt_file_path = corrupt_file_dir / "xxx.json"
     corrupt_file_path.write_text("BOOM")
 
-    [failure] = queue.ready()
-    assert failure.id == "queue.backend.retrieve"
-    assert failure.hint.startswith("invalid data: Unable to load item at path /")
+    [failure] = await queue.check_readable()
+    assert failure.id == "queue.backend.read"
+    assert failure.hint.startswith("check that parked event files are not corrupt")
 
 
 @pytest.mark.asyncio

tests/unit/jira/test_queue.py

@@ -479,7 +479,8 @@ def test_read_heartbeat_success(
                 "jira.all_projects_are_visible": "ok",
                 "jira.all_projects_have_permissions": "ok",
                 "jira.pandoc_install": "ok",
-                "queue.ready": "ok",
+                "queue.writable": "ok",
+                "queue.readable": "ok",
             },
             "details": {},
             "status": "ok",