perf(increment-approve): optimize OBS source report processing + development group filtering

openqabot/incrementapprover.py

@@ -7,6 +7,7 @@
 import os
 import re
 from collections import defaultdict
+from concurrent.futures import ThreadPoolExecutor
 from functools import lru_cache
 from itertools import chain
 from logging import getLogger
@@ -104,11 +105,7 @@ def get_regex_match(pattern: str, string: str) -> re.Match | None:
 
     def _filter_jobs(self, jobs: dict[str, dict[str, Any]]) -> dict[str, dict[str, Any]]:
         """Filter jobs within a state, removing those in devel groups."""
-        return {
-            name: {**info, "job_ids": ids}
-            for name, info in jobs.items()
-            if (ids := [i for i in info["job_ids"] if not self.is_in_devel_group(i)])
-        }
+        return {name: info for name, info in jobs.items() if not self.client.is_in_devel_group(info)}
 
     def _filter_results(self, results: OpenQAResults) -> OpenQAResults:
         """Remove jobs belonging to development groups from openQA results."""
@@ -117,17 +114,20 @@ def _filter_results(self, results: OpenQAResults) -> OpenQAResults:
     def request_openqa_job_results(self, params: ScheduleParams, info_str: str) -> OpenQAResults:
         """Fetch results from openQA for the specified scheduling parameters."""
         log.debug("Checking openQA job results for %s", info_str)
-        res = [
-            self.client.get_scheduled_product_stats({
+
+        def fetch(p: dict[str, Any]) -> OpenQAResult:
+            return self.client.get_scheduled_product_stats({
                 "distri": p["DISTRI"],
                 "version": p["VERSION"],
                 "flavor": p["FLAVOR"],
                 "arch": p["ARCH"],
                 "build": p["BUILD"],
                 "product": p.get("PRODUCT"),
             })
-            for p in params
-        ]
+
+        with ThreadPoolExecutor() as executor:
+            res = list(executor.map(fetch, params))
+
         log.debug("Job statistics:\n%s", pformat(res))
         return res
 
@@ -144,11 +144,6 @@ def check_openqa_jobs(results: OpenQAResults, build_info: BuildInfo, params: Sch
             return False
         return True
 
-    def is_in_devel_group(self, job_id: int) -> bool:
-        """Fetch job details and check if it belongs to a development group."""
-        job = self.client.get_single_job(job_id)
-        return self.client.is_in_devel_group(job) if job else False
-
     def evaluate_openqa_job_results(
         self,
         results: OpenQAResult,

openqabot/loader/sourcereport.py

@@ -6,6 +6,8 @@
 
 import tempfile
 from collections import defaultdict
+from concurrent.futures import ThreadPoolExecutor
+from functools import lru_cache
 from logging import getLogger
 from typing import Any
 from urllib.error import HTTPError
@@ -58,14 +60,20 @@ def parse_source_report(
         packages["noarch"].add(Package(b.get("package"), "", "", "", "noarch"))
 
 
+@lru_cache(maxsize=128)
+def get_repos_of_project(prj: str) -> list[Any]:
+    """Cache OBS repository lookups for a project."""
+    return list(osc.core.get_repos_of_project(config.settings.obs_url, prj=prj))
+
+
 def find_source_reports(project: str, package: str) -> list[str]:
     """Find source reports for a package in a project."""
-    repos = osc.core.get_repos_of_project(config.settings.obs_url, prj=project)
+    repos = get_repos_of_project(project)
     binaries = [
         osc.core.get_binarylist(config.settings.obs_url, prj=project, repo=repo.name, arch=repo.arch, package=package)
         for repo in repos
     ]
-    return [b for binary_list in binaries for b in binary_list if b.endswith("Source.report")]
+    return [b for binary_list in binaries for b in binary_list if b and b.endswith("Source.report")]
 
 
 def load_packages_from_source_report(
@@ -95,14 +103,26 @@ def compute_packages_of_request_from_source_report(
     """Compute the package diff of a request based on source reports."""
     repo_a: defaultdict[str, set[Package]] = defaultdict(set)
     repo_b: defaultdict[str, set[Package]] = defaultdict(set)
-    for action in request.actions:
-        log.debug("Checking action '%s' -> '%s' of request %s", action.src_project, action.tgt_project, request.id)
-        # add packages for target project (e.g. `SUSE:Products:SLE-Product-SLES:16.0:aarch64`), that is repo "A"
-        load_packages_from_source_report(
-            action, OBSBinary(action.tgt_project, action.src_package, "images", "local"), repo_a
-        )
-        # add packages for source project (e.g. `SUSE:SLFO:Products:SLES:16.0:TEST`), that is repo "B"
-        load_packages_from_source_report(
-            action, OBSBinary(action.src_project, action.src_package, "product", "local"), repo_b
-        )
+
+    def worker(action: Any, binary: OBSBinary) -> defaultdict[str, set[Package]]:  # noqa: ANN401
+        packages: defaultdict[str, set[Package]] = defaultdict(set)
+        load_packages_from_source_report(action, binary, packages)
+        return packages
+
+    tasks = [
+        *((a, OBSBinary(a.tgt_project, a.src_package, "images", "local")) for a in request.actions),
+        *((a, OBSBinary(a.src_project, a.src_package, "product", "local")) for a in request.actions),
+    ]
+
+    with ThreadPoolExecutor() as executor:
+        # Use executor.map to ensure exceptions are raised if they occur in threads
+        results = list(executor.map(lambda p: worker(*p), tasks))
+
+    # results contains first all "target" projects, then all "source" projects
+    num_actions = len(request.actions)
+    for i, res in enumerate(results):
+        target_repo = repo_a if i < num_actions else repo_b
+        for arch, pks in res.items():
+            target_repo[arch].update(pks)
+
     return RepoDiff.compute_diff_for_packages("product repo", repo_a, "TEST repo", repo_b)

tests/test_incrementapprover_helpers.py

@@ -84,9 +84,13 @@ def test_extra_builds_for_package_filtering(caplog: pytest.LogCaptureFixture) ->
 
 def test_filter_results(caplog: pytest.LogCaptureFixture, mocker: MockerFixture) -> None:
     approver = prepare_approver(caplog)
-    mocker.patch.object(approver.client, "is_in_devel_group", side_effect=lambda j: j.get("id") == 2)
-    mocker.patch.object(approver.client, "get_single_job", side_effect=lambda j: {"id": j})
+    mocker.patch.object(approver.client, "is_in_devel_group", side_effect=lambda j: j.get("group_id") == 9)
 
-    results = [{"passed": {"j1": {"job_ids": [1]}, "j2": {"job_ids": [1, 2]}}, "failed": {"j3": {"job_ids": [2]}}}]
-    expected = [{"passed": {"j1": {"job_ids": [1]}, "j2": {"job_ids": [1]}}}]
+    results = [
+        {
+            "passed": {"j1": {"job_ids": [1], "group_id": 1}, "j2": {"job_ids": [1, 2], "group_id": 9}},
+            "failed": {"j3": {"job_ids": [2], "group_id": 9}},
+        }
+    ]
+    expected = [{"passed": {"j1": {"job_ids": [1], "group_id": 1}}}]
     assert approver._filter_results(results) == expected  # noqa: SLF001

tests/test_incrementapprover_scenarios.py

@@ -509,7 +509,7 @@ def test_approval_if_failing_jobs_are_in_development_group(
     fake_openqa_url_job_stat: str,
     schedule: bool,  # noqa: FBT001
 ) -> None:
-    responses.add(responses.GET, fake_openqa_url_job_stat, json={"done": {"failed": {"job_ids": [123]}}})
+    responses.add(responses.GET, fake_openqa_url_job_stat, json={"done": {"failed": {"job_ids": [123], "group_id": 9}}})
     increment_approver = prepare_approver(caplog, schedule=schedule)
     # is_devel_group is called with group_id (int), unlike is_in_devel_group which takes a job dict
     increment_approver.client.get_single_job = mocker.Mock(return_value=_devel_job(123, result="failed"))
@@ -532,7 +532,7 @@ def test_approval_with_mixed_jobs_development_ignored(
     responses.add(
         responses.GET,
         fake_openqa_url_job_stat,
-        json={"done": {"failed": {"job_ids": [123]}, "passed": {"job_ids": [456]}}},
+        json={"done": {"failed": {"job_ids": [123], "group_id": 9}, "passed": {"job_ids": [456], "group_id": 1}}},
     )
     job_map = {123: _devel_job(123, result="failed"), 456: _prod_job(456, result="passed")}
     mock_osc_approve = mocker.patch("osc.core.change_review_state")
@@ -558,7 +558,10 @@ def test_approval_if_running_jobs_are_in_development_group(
     responses.add(
         responses.GET,
         fake_openqa_url_job_stat,
-        json={"running": {"some_job": {"job_ids": [123]}}, "done": {"passed": {"job_ids": [456]}}},
+        json={
+            "running": {"some_job": {"job_ids": [123], "group": "Development"}},
+            "done": {"passed": {"job_ids": [456], "group": "Production"}},
+        },
     )
     job_map = {123: _devel_job(123, state="running"), 456: _prod_job(456, state="done", result="passed")}
     mock_osc_approve = mocker.patch("osc.core.change_review_state")