ingester: cache list of files, speedup on large queue

Before: Every 5-second cycle calls os.scandir() on the 2M-entry directory.
Each scan enumerates all entries via readdir(), which is extremely slow
on a flat directory that large. Might take 20-30 seconds.

After:
1. scandir() runs once, caching all .json entries
2. Each cycle pops a chunk of up to INGEST_CYCLE_BATCH_SIZE (default 50,000) files from the cache and processes them
3. Re-scan only happens when the cache is fully drained
4. On scandir error, cache is cleared → forces re-scan next cycle

With 2M files: one scan instead of ~40 scans (2M / 50K chunks = 40 cycles of scan-free processing).
If each scandir of 2M entries takes ~30 seconds, that saves ~20 minutes of pure directory enumeration overhead.

The INGEST_CYCLE_BATCH_SIZE(50k default) is configurable via env var if you want to tune the chunk size.
Also note this fixed a latent bug in the old code where json_files would retain stale data from the previous
iteration if scandir threw an exception (the except: pass didn't reset it).

Signed-off-by: Denys Fedoryshchenko <denys.f@collabora.com>

Author: nuclearcat

backend/kernelCI_app/constants/ingester.py

@@ -43,6 +43,14 @@
 INGEST_FILES_BATCH_SIZE = int(os.environ.get("INGEST_FILES_BATCH_SIZE", 100))
 """Size of the batch of files to be queued. Default: 100"""
 
+try:
+    INGEST_CYCLE_BATCH_SIZE = int(os.environ.get("INGEST_CYCLE_BATCH_SIZE", "50000"))
+except (ValueError, TypeError):
+    logger.warning("Invalid INGEST_CYCLE_BATCH_SIZE, using default 50000")
+    INGEST_CYCLE_BATCH_SIZE = 50000
+"""Max files to process per cycle from cached scandir results.
+Avoids re-scanning huge directories between cycles. Default: 50000"""
+
 try:
     INGESTER_METRICS_PORT = int(os.environ.get("INGESTER_METRICS_PORT", 8002))
 except (ValueError, TypeError):

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

@@ -1,7 +1,6 @@
 import multiprocessing
 from multiprocessing.sharedctypes import Synchronized
 from multiprocessing.synchronize import Lock as ProcessLock
-from os import DirEntry
 import json
 import logging
 import os
@@ -487,7 +486,7 @@ def print_ingest_progress(
 
 
 def ingest_submissions_parallel(  # noqa: C901 - orchestrator with IO + multiprocessing
-    json_files: list[DirEntry[str]],
+    json_files: list[str],
     tree_names: dict[str, str],
     dirs: dict[INGESTER_DIRS, str],
     max_workers: int = 5,
@@ -504,17 +503,18 @@ def ingest_submissions_parallel(  # noqa: C901 - orchestrator with IO + multipro
     )
 
     batch = []
-    for file in json_files:
+    for file_path in json_files:
         try:
-            total_bytes += file.stat().st_size
-        except Exception:
-            pass
+            file_size = os.path.getsize(file_path)
+        except OSError:
+            file_size = 0
 
+        total_bytes += file_size
         batch.append(
             SubmissionFileMetadata(
-                path=file.path,
-                name=file.name,
-                size=file.stat().st_size,
+                path=file_path,
+                name=os.path.basename(file_path),
+                size=file_size,
             )
         )
 

backend/kernelCI_app/management/commands/monitor_submissions.py

@@ -10,6 +10,7 @@
     ingest_submissions_parallel,
 )
 from kernelCI_app.constants.ingester import (
+    INGEST_CYCLE_BATCH_SIZE,
     INGESTER_GRAFANA_LABEL,
     INGESTER_METRICS_PORT,
     PROMETHEUS_MULTIPROC_DIR,
@@ -50,6 +51,50 @@ def signal_handler(self, signum, frame):
         logger.info(f"Received signal {signum}, initiating graceful shutdown...")
         self.running = False
 
+    def _setup_prometheus(self):
+        if PROMETHEUS_MULTIPROC_DIR:
+            if os.path.exists(PROMETHEUS_MULTIPROC_DIR):
+                shutil.rmtree(PROMETHEUS_MULTIPROC_DIR)
+
+            os.makedirs(PROMETHEUS_MULTIPROC_DIR, exist_ok=True)
+            registry = CollectorRegistry()
+            multiprocess.MultiProcessCollector(registry)
+            start_http_server(INGESTER_METRICS_PORT, registry=registry)
+        else:
+            logger.warning(
+                "PROMETHEUS_MULTIPROC_DIR is not set, skipping Prometheus metrics"
+            )
+
+    def _scan_spool_dir(self, spool_dir: str) -> list[str]:
+        """Scan spool directory, returning only path strings to keep memory bounded."""
+        try:
+            with os.scandir(spool_dir) as it:
+                cached_paths = [
+                    entry.path
+                    for entry in it
+                    if entry.is_file() and entry.name.endswith(".json")
+                ]
+            ts = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
+            self.stdout.write(
+                f"[{ts}] Spool scan: {len(cached_paths)}" " .json files pending"
+            )
+            return cached_paths
+        except PermissionError:
+            logger.error(
+                "Permission denied scanning spool directory: %s",
+                spool_dir,
+                exc_info=True,
+            )
+            self.running = False
+            return []
+        except OSError:
+            logger.warning(
+                "Transient OS error scanning spool directory: %s",
+                spool_dir,
+                exc_info=True,
+            )
+            return []
+
     def add_arguments(self, parser):
         # TODO: add a way to set the folder by env var instead of by argument
         parser.add_argument(
@@ -89,18 +134,7 @@ def handle(
         signal.signal(signal.SIGTERM, self.signal_handler)
         signal.signal(signal.SIGINT, self.signal_handler)
 
-        if PROMETHEUS_MULTIPROC_DIR:
-            if os.path.exists(PROMETHEUS_MULTIPROC_DIR):
-                shutil.rmtree(PROMETHEUS_MULTIPROC_DIR)
-
-            os.makedirs(PROMETHEUS_MULTIPROC_DIR, exist_ok=True)
-            registry = CollectorRegistry()
-            multiprocess.MultiProcessCollector(registry)
-            start_http_server(INGESTER_METRICS_PORT, registry=registry)
-        else:
-            logger.warning(
-                "PROMETHEUS_MULTIPROC_DIR is not set, skipping Prometheus metrics"
-            )
+        self._setup_prometheus()
 
         dirs: dict[INGESTER_DIRS, str] = {
             "archive": os.path.join(spool_dir, "archive"),
@@ -120,32 +154,37 @@ def handle(
 
         self.stdout.write("Starting file monitoring... (Press Ctrl+C to stop)")
 
+        cached_files: list[str] = []
+        cache_pos = 0
+
         try:
             while self.running:
                 # TODO: retry failed files every x cycles
-                try:
-                    with os.scandir(spool_dir) as it:
-                        json_files = [
-                            entry
-                            for entry in it
-                            if entry.is_file() and entry.name.endswith(".json")
-                        ]
-                        ts = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
-                        self.stdout.write(
-                            f"[{ts}] Spool has {len(json_files)} .json files pending"
-                        )
-                except Exception:
-                    pass
-
-                QUEUE_SIZE_GAUGE.labels(INGESTER_GRAFANA_LABEL).set(len(json_files))
-
-                if len(json_files) > 0:
+
+                # Only re-scan directory when cache is depleted
+                if cache_pos >= len(cached_files):
+                    cached_files = self._scan_spool_dir(spool_dir)
+                    cache_pos = 0
+
+                remaining = len(cached_files) - cache_pos
+                QUEUE_SIZE_GAUGE.labels(INGESTER_GRAFANA_LABEL).set(remaining)
+
+                if remaining > 0:
+                    end = min(cache_pos + INGEST_CYCLE_BATCH_SIZE, len(cached_files))
+                    batch = cached_files[cache_pos:end]
+                    cache_pos = end
+                    self.stdout.write(
+                        f"Processing {len(batch)} files"
+                        f" ({len(cached_files) - cache_pos}"
+                        " remaining in cache)"
+                    )
                     ingest_submissions_parallel(
-                        json_files,
+                        batch,
                         tree_names,
                         dirs,
                         max_workers,
                     )
+
                 cache_logs_maintenance()
 
                 time.sleep(interval)

backend/kernelCI_app/tests/performanceTests/test_ingest_perf.py

@@ -34,10 +34,12 @@
 FILE_SUBSETS = [100, 300, 500]
 
 
-def _load_submission_files(dir_path: str) -> list[os.DirEntry[str]]:
+def _load_submission_files(dir_path: str) -> list[str]:
     with os.scandir(dir_path) as it:
         return [
-            entry for entry in it if entry.is_file() and entry.name.endswith(".json")
+            entry.path
+            for entry in it
+            if entry.is_file() and entry.name.endswith(".json")
         ]
 
 
@@ -102,9 +104,7 @@ def cleanup_submission_files():
     _delete_json_files()
 
 
-def _get_file_subset(
-    files: list[os.DirEntry[str]], subset: int
-) -> list[os.DirEntry[str]]:
+def _get_file_subset(files: list[str], subset: int) -> list[str]:
     """Get a subset of files based on the subset type."""
     return files[:subset] if len(files) >= subset else files
 
@@ -212,14 +212,12 @@ def test_prepare_file_data(
 
     assert len(files) > 0, "No submissions found"
 
-    def prepare_files(
-        files: list[os.DirEntry[str]], trees_names: dict[str, str]
-    ) -> None:
-        for file in files:
+    def prepare_files(files: list[str], trees_names: dict[str, str]) -> None:
+        for file_path in files:
             file_metadata: SubmissionFileMetadata = {
-                "path": file.path,
-                "name": file.name,
-                "size": file.stat().st_size,
+                "path": file_path,
+                "name": os.path.basename(file_path),
+                "size": os.path.getsize(file_path),
             }
 
             prepare_file_data(file_metadata, trees_names)
@@ -242,7 +240,7 @@ def prepare_files(
 
 
 def _prepare_buffers(
-    files: list[os.DirEntry[str]], trees_names: dict[str, str]
+    files: list[str], trees_names: dict[str, str]
 ) -> tuple[list, dict[str, list[Any]]]:
     """
     Prepare object buffers from submission files.
@@ -260,11 +258,11 @@ def _prepare_buffers(
     }
     buffer_files: set[tuple[str, str]] = set()
 
-    for file in files:
+    for file_path in files:
         file_metadata: SubmissionFileMetadata = {
-            "path": file.path,
-            "name": file.name,
-            "size": file.stat().st_size,
+            "path": file_path,
+            "name": os.path.basename(file_path),
+            "size": os.path.getsize(file_path),
         }
 
         data, metadata = prepare_file_data(file_metadata, trees_names)
@@ -279,7 +277,7 @@ 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))
+        buffer_files.add((os.path.basename(file_path), file_path))
 
     return [], {
         "issues_buf": objects_buffers["issues"],
@@ -365,11 +363,11 @@ def test_build_instances_perf(benchmark, cleanup_submission_files, file_subset):
     assert len(files) > 0, "No submissions found"
 
     data_list = []
-    for file in files:
+    for file_path in files:
         file_metadata: SubmissionFileMetadata = {
-            "path": file.path,
-            "name": file.name,
-            "size": file.stat().st_size,
+            "path": file_path,
+            "name": os.path.basename(file_path),
+            "size": os.path.getsize(file_path),
         }
 
         data, _ = prepare_file_data(file_metadata, trees_names)

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

@@ -790,18 +790,19 @@ class TestIngestSubmissionsParallel:
     # Test cases:
     # - successful ingestion
 
-    mock_file1 = MagicMock()
-    mock_file1.name = SUBMISSION_FILENAME_MOCK
-    mock_file1.stat.return_value.st_size = 1000
+    FILE1_SIZE = 1000
+    FILE2_SIZE = 2000
 
     @patch("kernelCI_app.management.commands.helpers.kcidbng_ingester.out")
     @patch("multiprocessing.Process")
     @patch("multiprocessing.Queue")
     @patch("multiprocessing.Value")
     @patch("time.sleep")
     @patch("time.time", side_effect=TIME_MOCK)
+    @patch("os.path.getsize")
     def test_ingest_submissions_parallel_success(
         self,
+        mock_getsize,
         mock_time,
         mock_sleep,
         mock_value,
@@ -810,19 +811,19 @@ def test_ingest_submissions_parallel_success(
         mock_out,
     ):
         """Test successful parallel ingestion."""
+        file1_path = SUBMISSION_FILEPATH_MOCK + SUBMISSION_FILENAME_MOCK
+        file2_path = SUBMISSION_FILEPATH_MOCK + "file2.json"
+
+        file_sizes = {file1_path: self.FILE1_SIZE, file2_path: self.FILE2_SIZE}
+        mock_getsize.side_effect = lambda p: file_sizes[p]
+
         mock_queue = MagicMock()
         mock_queue_cls.return_value = mock_queue
 
         mock_queue.empty.return_value = True
         mock_queue.qsize.return_value = 0
 
-        self.mock_file1.path = SUBMISSION_FILEPATH_MOCK + SUBMISSION_FILENAME_MOCK
-        mock_file2 = MagicMock()
-        mock_file2.name = "file2.json"
-        mock_file2.path = SUBMISSION_FILEPATH_MOCK + "file2.json"
-        mock_file2.stat.return_value.st_size = 2000
-
-        json_files = [self.mock_file1, mock_file2]
+        json_files = [file1_path, file2_path]
 
         mock_ok = MagicMock()
         mock_ok.value = 1
@@ -852,10 +853,7 @@ def test_ingest_submissions_parallel_success(
         stat_fail = 1
         total_elapsed = 1
 
-        total_bytes = (
-            self.mock_file1.stat.return_value.st_size
-            + mock_file2.stat.return_value.st_size
-        )
+        total_bytes = self.FILE1_SIZE + self.FILE2_SIZE
         mb = total_bytes / (1024 * 1024)
 
         # We verify only the final call