add concurrent processing with deadlock workaround and progress bar fixes

- add run_concurrently() helper with multiprocessing.Pool fallback for python < 3.12 (CPython #105829)
- add CONCURRENT_PROCESSING and CONCURRENT_WORKERS settings
- add max_workers param and serial fallback to all errata sources
- switch osv.dev enrichment to two-phase fetch/parse with connection pooling
- add exception handling for threadpool futures in enrich_errata
- switch arch to ThreadPoolExecutor, default serial to avoid rate limiting
- add null guards for arch errata fetch failures
- disable tqdm TMonitor thread with monitor_interval = 0
- add clear_forked_pbar to wrapper functions for forked workers
- add prefetch_related to alma modules, batch module adds for rocky
- add mininterval=0.5 to tqdm progress bars
- move function-level imports to top-level in errata sources

Author: furlongm

errata/models.py

@@ -25,7 +25,7 @@
 from security.models import CVE, Reference
 from security.utils import get_or_create_cve, get_or_create_reference
 from util import get_url
-from util.logging import error_message
+from util.logging import debug_message, error_message
 
 
 class Erratum(models.Model):
@@ -91,15 +91,27 @@ def _mark_updates_security(self, updates):
                     error_message(text=e)
                     update.delete()
 
-    def fetch_osv_dev_data(self):
+    def fetch_osv_dev_data(self, session=None):
+        """ Fetch osv.dev JSON for this erratum. Returns parsed JSON or None.
+        """
         osv_dev_url = f'https://api.osv.dev/v1/vulns/{self.name}'
-        res = get_url(osv_dev_url)
+        debug_message(text=f'Trying {osv_dev_url}')
+        try:
+            if session:
+                res = session.get(osv_dev_url, timeout=30)
+            else:
+                res = get_url(osv_dev_url)
+                if res is None:
+                    error_message(text=f'No response - Skipping {self.name} - {osv_dev_url}')
+                    return None
+        except Exception as e:
+            error_message(text=f'Error fetching {osv_dev_url}: {e}')
+            return None
+        debug_message(text=f'{res.status_code}: {res.headers}')
         if res.status_code == 404:
-            error_message(text=f'404 - Skipping {self.name} - {osv_dev_url}')
-            return
+            return None
         data = res.content
-        osv_dev_json = json.loads(data)
-        self.parse_osv_dev_data(osv_dev_json)
+        return json.loads(data)
 
     def parse_osv_dev_data(self, osv_dev_json):
         from django.db.models import Q

errata/sources/distros/alma.py

@@ -14,19 +14,21 @@
 # You should have received a copy of the GNU General Public License
 # along with Patchman. If not, see <http://www.gnu.org/licenses/>
 
-import concurrent.futures
 import json
 
-from django.db import connections
-
-from operatingsystems.utils import get_or_create_osrelease
+from errata.utils import get_or_create_erratum
+from modules.utils import get_matching_modules
+from operatingsystems.utils import (
+    get_or_create_osrelease, normalize_el_osrelease,
+)
 from packages.models import Package
 from packages.utils import get_or_create_package, parse_package_string
 from patchman.signals import pbar_start, pbar_update
-from util import fetch_content, get_setting_of_type, get_url
+from util import fetch_content, get_setting_of_type, get_url, run_concurrently
+from util.logging import clear_forked_pbar
 
 
-def update_alma_errata(concurrent_processing=True):
+def update_alma_errata(concurrent_processing=True, max_workers=25):
     """ Update Alma Linux advisories from errata.almalinux.org:
            https://errata.almalinux.org/8/errata.full.json
            https://errata.almalinux.org/9/errata.full.json
@@ -40,7 +42,7 @@ def update_alma_errata(concurrent_processing=True):
     )
     for release in alma_releases:
         advisories = fetch_alma_advisories(release)
-        process_alma_errata(release, advisories, concurrent_processing)
+        process_alma_errata(release, advisories, concurrent_processing, max_workers)
 
 
 def fetch_alma_advisories(release):
@@ -54,11 +56,11 @@ def fetch_alma_advisories(release):
     return advisories
 
 
-def process_alma_errata(release, advisories, concurrent_processing):
+def process_alma_errata(release, advisories, concurrent_processing, max_workers=25):
     """ Process Alma Linux Errata
     """
     if concurrent_processing:
-        process_alma_errata_concurrently(release, advisories)
+        process_alma_errata_concurrently(release, advisories, max_workers)
     else:
         process_alma_errata_serially(release, advisories)
 
@@ -73,24 +75,24 @@ def process_alma_errata_serially(release, advisories):
         pbar_update.send(sender=None, index=i + 1)
 
 
-def process_alma_errata_concurrently(release, advisories):
+def process_alma_errata_concurrently(release, advisories, max_workers=25):
     """ Process Alma Linux Errata concurrently
     """
-    connections.close_all()
     elen = len(advisories)
     pbar_start.send(sender=None, ptext=f'Processing {elen} Alma {release} Errata', plen=elen)
-    i = 0
-    with concurrent.futures.ProcessPoolExecutor(max_workers=25) as executor:
-        futures = [executor.submit(process_alma_erratum, release, advisory) for advisory in advisories]
-        for future in concurrent.futures.as_completed(futures):
-            i += 1
-            pbar_update.send(sender=None, index=i + 1)
+    args = [(release, advisory) for advisory in advisories]
+    for i, _ in enumerate(run_concurrently(_process_alma_erratum_wrapper, args, max_workers)):
+        pbar_update.send(sender=None, index=i + 1)
+
+
+def _process_alma_erratum_wrapper(args):
+    clear_forked_pbar()
+    return process_alma_erratum(*args)
 
 
 def process_alma_erratum(release, advisory):
     """ Process a single Alma Linux Erratum
     """
-    from errata.utils import get_or_create_erratum
     erratum_name = advisory.get('id')
     issue_date = advisory.get('issued_date')
     synopsis = advisory.get('title')
@@ -110,7 +112,6 @@ def process_alma_erratum(release, advisory):
 def add_alma_erratum_osreleases(e, release):
     """ Update OS Release for Alma Linux errata
     """
-    from operatingsystems.utils import normalize_el_osrelease
     osrelease_name = normalize_el_osrelease(f'Alma Linux {release}')
     osrelease = get_or_create_osrelease(name=osrelease_name)
     e.osreleases.add(osrelease)
@@ -150,7 +151,6 @@ def add_alma_erratum_packages(e, advisory):
 def add_alma_erratum_modules(e, advisory):
     """ Parse and add modules for Alma Linux errata
     """
-    from modules.utils import get_matching_modules
     fixed_packages = set()
     modules = advisory.get('modules')
     for module in modules:
@@ -160,8 +160,7 @@ def add_alma_erratum_modules(e, advisory):
         stream = module.get('stream')
         version = module.get('version')
         matching_modules = get_matching_modules(name, stream, version, context, arch)
-        for match in matching_modules:
+        for match in matching_modules.prefetch_related('packages'):
             for fixed_package in match.packages.all():
-                match.packages.add(fixed_package)
                 fixed_packages.add(fixed_package)
     e.add_fixed_packages(fixed_packages)

errata/sources/distros/arch.py

@@ -17,25 +17,25 @@
 import concurrent.futures
 import json
 
-from django.db import connections
-
+from errata.utils import get_or_create_erratum
 from operatingsystems.utils import get_or_create_osrelease
 from packages.models import Package
 from packages.utils import (
     find_evr, get_matching_packages, get_or_create_package,
 )
 from patchman.signals import pbar_start, pbar_update
 from util import fetch_content, get_url
-from util.logging import error_message
+from util.logging import clear_forked_pbar, error_message
 
 
-def update_arch_errata(concurrent_processing=False):
+def update_arch_errata(concurrent_processing=False, max_workers=25):
     """ Update Arch Linux Errata from the following sources:
         https://security.archlinux.org/advisories.json
     """
     add_arch_linux_osrelease()
     advisories = fetch_arch_errata()
-    parse_arch_errata(advisories, concurrent_processing)
+    if advisories:
+        parse_arch_errata(advisories, concurrent_processing, max_workers)
 
 
 def fetch_arch_errata():
@@ -44,14 +44,16 @@ def fetch_arch_errata():
     """
     res = get_url('https://security.archlinux.org/advisories.json')
     advisories = fetch_content(res, 'Fetching Arch Advisories')
+    if advisories is None:
+        return None
     return json.loads(advisories)
 
 
-def parse_arch_errata(advisories, concurrent_processing):
+def parse_arch_errata(advisories, concurrent_processing, max_workers=25):
     """ Parse Arch Linux Errata Advisories
     """
     if concurrent_processing:
-        parse_arch_errata_concurrently(advisories)
+        parse_arch_errata_concurrently(advisories, max_workers)
     else:
         parse_arch_errata_serially(advisories)
 
@@ -67,15 +69,14 @@ def parse_arch_errata_serially(advisories):
         pbar_update.send(sender=None, index=i + 1)
 
 
-def parse_arch_errata_concurrently(advisories):
+def parse_arch_errata_concurrently(advisories, max_workers=25):
     """ Parse Arch Linux Errata Advisories concurrently
     """
     osrelease = get_or_create_osrelease(name='Arch Linux')
-    connections.close_all()
     elen = len(advisories)
     pbar_start.send(sender=None, ptext=f'Processing {elen} Arch Advisories', plen=elen)
     i = 0
-    with concurrent.futures.ProcessPoolExecutor(max_workers=25) as executor:
+    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
         futures = [executor.submit(process_arch_erratum, advisory, osrelease) for advisory in advisories]
         for future in concurrent.futures.as_completed(futures):
             i += 1
@@ -85,7 +86,7 @@ def parse_arch_errata_concurrently(advisories):
 def process_arch_erratum(advisory, osrelease):
     """ Process a single Arch Linux Erratum
     """
-    from errata.utils import get_or_create_erratum
+    clear_forked_pbar()
     try:
         name = advisory.get('name')
         issue_date = advisory.get('date')
@@ -121,6 +122,8 @@ def add_arch_erratum_references(e, advisory):
     e.add_reference('ASA', url)
     raw_url = f'{url}/raw'
     res = get_url(raw_url)
+    if res is None:
+        return
     data = res.content
     parse_arch_erratum_raw(e, data.decode())
 
@@ -152,6 +155,8 @@ def add_arch_erratum_packages(e, advisory):
     group_id = advisory.get('group')
     group_url = f'https://security.archlinux.org/group/{group_id}.json'
     res = get_url(group_url)
+    if res is None:
+        return
     data = res.content
     group = json.loads(data)
     packages = group.get('packages')