feat(ingester): add pending_retry folder (#1678)

* feat(ingester): add pending_retry folder

Fixes issues with file counter since the files were moved directly to archive before their data was processed, since their data was moved to the queue directly.

Part of #1376

* fix: ingester perf tests

Also adjusts performance tests docs

Author: MarceloRobert

backend/kernelCI_app/management/commands/helpers/file_utils.py

@@ -45,6 +45,8 @@ def verify_dir(dir: str) -> None:
 def verify_spool_dirs(spool_dir: str) -> None:
     failed_dir = os.path.join(spool_dir, "failed")
     archive_dir = os.path.join(spool_dir, "archive")
+    pending_retry_dir = os.path.join(spool_dir, "pending_retry")
     verify_dir(spool_dir)
     verify_dir(failed_dir)
     verify_dir(archive_dir)
+    verify_dir(pending_retry_dir)

backend/kernelCI_app/management/commands/helpers/kcidbng_ingester.py

@@ -6,6 +6,7 @@
 import logging
 import os
 from queue import Queue, Empty
+from typing_extensions import Literal
 from kernelCI_app.constants.ingester import (
     CONVERT_LOG_EXCERPT,
     INGEST_BATCH_SIZE,
@@ -38,13 +39,7 @@
 
 from prometheus_client import Counter
 
-
-class SubmissionMetadata(TypedDict):
-    filename: str
-    full_filename: str
-    fsize: int | None
-    processing_time: float | None
-    error: str | None
+type INGESTER_DIRS = Literal["archive", "failed", "pending_retry"]
 
 
 class SubmissionFileMetadata(TypedDict):
@@ -177,6 +172,9 @@ def flush_buffers(
     builds_buf: list[Builds],
     tests_buf: list[Tests],
     incidents_buf: list[Incidents],
+    buffer_files: set[tuple[str, str]],
+    archive_dir: str,
+    pending_retry_dir: str,
 ) -> None:
     """
     Consumes the list of objects and tries to insert them into the database.
@@ -206,8 +204,17 @@ def flush_buffers(
                 checkouts_instances=checkouts_buf,
                 tests_instances=tests_buf,
             )
+        for filename, filepath in buffer_files:
+            os.rename(filepath, os.path.join(archive_dir, filename))
     except Exception as e:
         logger.error("Error during buffer flush: %s", e)
+        try:
+            for filename, filepath in buffer_files:
+                os.rename(filepath, os.path.join(pending_retry_dir, filename))
+            out("Moved %d files to pending retry directory" % len(buffer_files))
+        except OSError as oe:
+            logger.error("OS error during buffer file pending retry move: %s", oe)
+            logger.error("Removing files from buffer set, they should be retried")
     finally:
         flush_dur = time.time() - flush_start
         rate = total / flush_dur if flush_dur > 0 else 0.0
@@ -230,10 +237,16 @@ def flush_buffers(
         builds_buf.clear()
         tests_buf.clear()
         incidents_buf.clear()
+        buffer_files.clear()
 
 
 # TODO: lower the complexity of this function
-def db_worker(stop_event: EventClass, db_queue: Queue) -> None:  # noqa: C901
+def db_worker(  # noqa: C901
+    stop_event: EventClass,
+    db_queue: Queue,
+    archive_dir: str,
+    pending_retry_dir: str,
+) -> None:
     """
     Worker process that processes the database queue.
     This is the only process that interacts with the database.
@@ -249,6 +262,7 @@ def db_worker(stop_event: EventClass, db_queue: Queue) -> None:  # noqa: C901
     builds_buf: list[Builds] = []
     tests_buf: list[Tests] = []
     incidents_buf: list[Incidents] = []
+    buffer_files: set[tuple[str, str]] = set()
 
     last_flush_ts = time.time()
 
@@ -268,8 +282,9 @@ def buffered_total() -> int:
                 db_queue.task_done()
                 break
             try:
-                filename, inst = item
+                filename, filepath, inst = item
                 if inst is not None:
+                    buffer_files.add((filename, filepath))
                     issues_buf.extend(inst["issues"])
                     checkouts_buf.extend(inst["checkouts"])
                     builds_buf.extend(inst["builds"])
@@ -283,6 +298,9 @@ def buffered_total() -> int:
                         builds_buf=builds_buf,
                         tests_buf=tests_buf,
                         incidents_buf=incidents_buf,
+                        archive_dir=archive_dir,
+                        pending_retry_dir=pending_retry_dir,
+                        buffer_files=buffer_files,
                     )
                     last_flush_ts = time.time()
 
@@ -322,6 +340,9 @@ def buffered_total() -> int:
                     builds_buf=builds_buf,
                     tests_buf=tests_buf,
                     incidents_buf=incidents_buf,
+                    archive_dir=archive_dir,
+                    pending_retry_dir=pending_retry_dir,
+                    buffer_files=buffer_files,
                 )
                 last_flush_ts = time.time()
             continue
@@ -335,6 +356,9 @@ def buffered_total() -> int:
         builds_buf=builds_buf,
         tests_buf=tests_buf,
         incidents_buf=incidents_buf,
+        archive_dir=archive_dir,
+        pending_retry_dir=pending_retry_dir,
+        buffer_files=buffer_files,
     )
 
 
@@ -351,7 +375,6 @@ def process_file(
     file: SubmissionFileMetadata,
     tree_names: dict[str, str],
     failed_dir: str,
-    archive_dir: str,
     db_queue: Queue,
 ) -> bool:
     """
@@ -374,25 +397,21 @@ def process_file(
         return True
 
     db_queue.put(
-        (file["name"], build_instances_from_submission(data, MAP_TABLENAMES_TO_COUNTER))
+        (
+            file["name"],
+            file["path"],
+            build_instances_from_submission(data, MAP_TABLENAMES_TO_COUNTER),
+        )
     )
     FILES_INGESTER_COUNTER.labels(INGESTER_GRAFANA_LABEL).inc()
 
-    # Archive the file after queuing (we can do this optimistically)
-    try:
-        os.rename(file["path"], os.path.join(archive_dir, file["name"]))
-    except Exception as e:
-        logger.error("Error archiving file %s: %s", file["name"], e)
-        return False
-
     return True
 
 
 def ingest_submissions_parallel(  # noqa: C901 - orchestrator with IO + threading
     json_files: list[DirEntry[str]],
     tree_names: dict[str, str],
-    archive_dir: str,
-    failed_dir: str,
+    dirs: dict[INGESTER_DIRS, str],
     max_workers: int = 5,
 ) -> None:
     """
@@ -423,7 +442,15 @@ def ingest_submissions_parallel(  # noqa: C901 - orchestrator with IO + threadin
     # Start database worker process
     # This process will constantly consume the db_queue and send data to the database
     stop_event = multiprocessing.Event()
-    db_process = multiprocessing.Process(target=db_worker, args=(stop_event, db_queue))
+    db_process = multiprocessing.Process(
+        target=db_worker,
+        args=(
+            stop_event,
+            db_queue,
+            dirs["archive"],
+            dirs["pending_retry"],
+        ),
+    )
     db_process.start()
 
     stat_ok = 0
@@ -443,8 +470,7 @@ def ingest_submissions_parallel(  # noqa: C901 - orchestrator with IO + threadin
                     process_file,
                     {"path": file.path, "name": file.name, "size": file.stat().st_size},
                     tree_names,
-                    failed_dir,
-                    archive_dir,
+                    dirs["failed"],
                     db_queue,
                 ): file.name
                 for file in json_files

backend/kernelCI_app/management/commands/monitor_submissions.py

@@ -5,6 +5,7 @@
 import time
 import os
 from kernelCI_app.management.commands.helpers.kcidbng_ingester import (
+    INGESTER_DIRS,
     ingest_submissions_parallel,
 )
 from kernelCI_app.constants.ingester import (
@@ -91,12 +92,16 @@ def handle(
                 "PROMETHEUS_MULTIPROC_DIR is not set, skipping Prometheus metrics"
             )
 
-        archive_dir = os.path.join(spool_dir, "archive")
-        failed_dir = os.path.join(spool_dir, "failed")
+        dirs: dict[INGESTER_DIRS, str] = {
+            "archive": os.path.join(spool_dir, "archive"),
+            "failed": os.path.join(spool_dir, "failed"),
+            "pending_retry": os.path.join(spool_dir, "pending_retry"),
+        }
 
         self.stdout.write(f"Monitoring folder: {spool_dir}")
-        self.stdout.write(f"Archive directory: {archive_dir}")
-        self.stdout.write(f"Failed directory: {failed_dir}")
+        self.stdout.write(f"Archive directory: {dirs['archive']}")
+        self.stdout.write(f"Failed directory: {dirs['failed']}")
+        self.stdout.write(f"Pending retry directory: {dirs['pending_retry']}")
         self.stdout.write(f"Check interval: {interval} seconds")
         self.stdout.write(f"Using {max_workers} workers")
 
@@ -126,7 +131,10 @@ def handle(
 
                 if len(json_files) > 0:
                     ingest_submissions_parallel(
-                        json_files, tree_names, archive_dir, failed_dir, max_workers
+                        json_files,
+                        tree_names,
+                        dirs,
+                        max_workers,
                     )
                 cache_logs_maintenance()
 

backend/kernelCI_app/tests/performanceTests/test_ingest_perf.py

@@ -125,8 +125,11 @@ def test_ingest_perf_workers(
         args=(
             files,
             trees_names,
-            os.path.join(SUBMISSIONS_DIR, "archive"),
-            os.path.join(SUBMISSIONS_DIR, "failed"),
+            {
+                "archive": os.path.join(SUBMISSIONS_DIR, "archive"),
+                "failed": os.path.join(SUBMISSIONS_DIR, "failed"),
+                "pending_retry": os.path.join(SUBMISSIONS_DIR, "pending_retry"),
+            },
             max_workers,
         ),
         rounds=5,
@@ -158,8 +161,11 @@ def test_ingest_perf_file_count(
         args=(
             files,
             trees_names,
-            os.path.join(SUBMISSIONS_DIR, "archive"),
-            os.path.join(SUBMISSIONS_DIR, "failed"),
+            {
+                "archive": os.path.join(SUBMISSIONS_DIR, "archive"),
+                "failed": os.path.join(SUBMISSIONS_DIR, "failed"),
+                "pending_retry": os.path.join(SUBMISSIONS_DIR, "pending_retry"),
+            },
             3,
         ),
         rounds=5,
@@ -252,6 +258,7 @@ def _prepare_buffers(
         "tests": [],
         "incidents": [],
     }
+    buffer_files: set[tuple[str, str]] = set()
 
     for file in files:
         file_metadata: SubmissionFileMetadata = {
@@ -272,13 +279,17 @@ def _prepare_buffers(
         objects_buffers["builds"].extend(instances["builds"])
         objects_buffers["tests"].extend(instances["tests"])
         objects_buffers["incidents"].extend(instances["incidents"])
+        buffer_files.add((file.name, file.path))
 
     return [], {
         "issues_buf": objects_buffers["issues"],
         "checkouts_buf": objects_buffers["checkouts"],
         "builds_buf": objects_buffers["builds"],
         "tests_buf": objects_buffers["tests"],
         "incidents_buf": objects_buffers["incidents"],
+        "buffer_files": buffer_files,
+        "archive_dir": os.path.join(SUBMISSIONS_DIR, "archive"),
+        "pending_retry_dir": os.path.join(SUBMISSIONS_DIR, "pending_retry"),
     }
 
 
@@ -325,8 +336,11 @@ def test_ingest_parallel_batch_size(
             args=(
                 files,
                 trees_names,
-                os.path.join(SUBMISSIONS_DIR, "archive"),
-                os.path.join(SUBMISSIONS_DIR, "failed"),
+                {
+                    "archive": os.path.join(SUBMISSIONS_DIR, "archive"),
+                    "failed": os.path.join(SUBMISSIONS_DIR, "failed"),
+                    "pending_retry": os.path.join(SUBMISSIONS_DIR, "pending_retry"),
+                },
                 1,
             ),
             rounds=5,

backend/kernelCI_app/tests/unitTests/commands/monitorSubmissions/file_utils_test.py

@@ -9,6 +9,7 @@
     verify_spool_dirs,
 )
 from kernelCI_app.tests.unitTests.helpers.fixtures.file_utils_data import (
+    PENDING_RETRY_SPOOL_SUBDIR,
     TREES_PATH_TESTING,
     BASE_TREES_FILE,
     EXPECTED_PARSED_TREES_FILE,
@@ -242,13 +243,17 @@ def test_verify_spool_dirs_success(self, mock_verify_dir):
         """Test verify_spool_dirs with successful directory verification."""
         joined_fail_dir = "/".join([SPOOL_DIR_TESTING, FAIL_SPOOL_SUBDIR])
         joined_archive_dir = "/".join([SPOOL_DIR_TESTING, ARCHIVE_SPOOL_SUBDIR])
+        joined_pending_retry_dir = "/".join(
+            [SPOOL_DIR_TESTING, PENDING_RETRY_SPOOL_SUBDIR]
+        )
 
         verify_spool_dirs(SPOOL_DIR_TESTING)
 
-        assert mock_verify_dir.call_count == 3
+        assert mock_verify_dir.call_count == 4
         mock_verify_dir.assert_any_call(SPOOL_DIR_TESTING)
         mock_verify_dir.assert_any_call(joined_fail_dir)
         mock_verify_dir.assert_any_call(joined_archive_dir)
+        mock_verify_dir.assert_any_call(joined_pending_retry_dir)
 
     @patch("kernelCI_app.management.commands.helpers.file_utils.verify_dir")
     def test_verify_spool_dirs_verify_spool_dir_fails(self, mock_verify_dir):