Ref #967: Add queue parsing check in heartbeat (#968)

Add queue parsing check in heartbeat

Author: leplatrem

.env.example

@@ -5,7 +5,7 @@ LOG_LEVEL=debug
 APP_RELOAD=True
 APP_DEBUG=True
 JBI_API_KEY="fake_api_key"
-DL_QUEUE_DSN=file:///tmp
+DL_QUEUE_DSN=file:///tmp/dlqueue
 DL_QUEUE_CONSTANT_RETRY=true
 DL_QUEUE_RETRY_TIMEOUT_DAYS=7
 

jbi/queue.py

@@ -22,6 +22,7 @@
       item and parse it as an item
 """
 
+import asyncio
 import logging
 import tempfile
 import traceback
@@ -30,7 +31,7 @@
 from functools import lru_cache
 from json import JSONDecodeError
 from pathlib import Path
-from typing import AsyncIterator, List, Optional
+from typing import Any, AsyncIterator, List, Optional
 from urllib.parse import ParseResult, urlparse
 
 import dockerflow.checks
@@ -42,6 +43,10 @@
 logger = logging.getLogger(__name__)
 
 
+async def async_iter(iter: AsyncIterator[Any]) -> list[Any]:
+    return [item async for item in iter]
+
+
 class QueueItemRetrievalError(Exception):
     pass
 
@@ -160,6 +165,9 @@ def __init__(self, location):
         self.location = Path(location)
         self.location.mkdir(parents=True, exist_ok=True)
 
+    def __repr__(self) -> str:
+        return f"FileBackend({self.location})"
+
     def ping(self):
         try:
             with tempfile.TemporaryDirectory(dir=self.location) as temp_dir:
@@ -222,7 +230,7 @@ async def get(self, bug_id: int) -> AsyncIterator[QueueItem]:
                 yield QueueItem.parse_file(path)
             except (JSONDecodeError, ValidationError) as e:
                 raise QueueItemRetrievalError(
-                    "Unable to load item at path %s from queue", str(path)
+                    f"Unable to load item at path {path} from queue"
                 ) from e
 
     async def get_all(self) -> dict[int, AsyncIterator[QueueItem]]:
@@ -253,15 +261,31 @@ def ready(self) -> list[dockerflow.checks.CheckMessage]:
         TODO: Convert to an async method when Dockerflow's FastAPI integration
         can run check asynchronously
         """
-
+        results = []
         ping_result = self.backend.ping()
         if ping_result is False:
-            return [
+            results.append(
                 dockerflow.checks.Error(
-                    f"queue with f{str(self.backend)} backend unavailable"
+                    f"queue with {str(self.backend)} unavailable",
+                    hint="with FileBackend, check that folder is writable",
+                    id="queue.backend.ping",
                 )
-            ]
-        return []
+            )
+
+        try:
+            bugs_items = asyncio.run(self.retrieve())
+            for items in bugs_items.values():
+                asyncio.run(async_iter(items))
+        except Exception as exc:
+            results.append(
+                dockerflow.checks.Error(
+                    f"queue with {str(self.backend)} cannot be retrieved",
+                    hint=f"invalid data: {exc}",
+                    id="queue.backend.retrieve",
+                )
+            )
+
+        return results
 
     async def postpone(self, payload: bugzilla.WebhookRequest) -> None:
         """

tests/unit/jira/test_queue.py

@@ -242,6 +242,28 @@ async def test_get_payload_doesnt_match_schema(
         await anext(items)
 
 
+def test_ready_ok(queue: DeadLetterQueue):
+    assert queue.ready() == []
+
+
+def test_ready_not_writable(queue: DeadLetterQueue, tmp_path):
+    queue.backend = FileBackend(tmp_path)
+    tmp_path.chmod(0o400)  # set to readonly
+    [failure] = queue.ready()
+    assert failure.id == "queue.backend.ping"
+
+
+def test_ready_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 /")
+
+
 @pytest.mark.asyncio
 async def test_postpone(queue: DeadLetterQueue, webhook_request_factory):
     webhook_payload = webhook_request_factory()