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

Author: furlongm

errata/models.py

@@ -91,15 +91,16 @@ 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)
+        res = get_url(osv_dev_url, session=session)
+        if res is None:
+            return None
         if res.status_code == 404:
-            error_message(text=f'404 - Skipping {self.name} - {osv_dev_url}')
-            return
-        data = res.content
-        osv_dev_json = json.loads(data)
-        self.parse_osv_dev_data(osv_dev_json)
+            return None
+        return json.loads(res.content)
 
     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')

errata/sources/distros/debian.py

@@ -14,27 +14,28 @@
 # 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 csv
 import re
 from datetime import datetime
 from io import StringIO
 
 from debian.deb822 import Dsc
-from django.db import connections
 
+from errata.utils import get_or_create_erratum
 from operatingsystems.models import OSRelease
 from operatingsystems.utils import get_or_create_osrelease
 from packages.models import Package
 from packages.utils import find_evr, get_or_create_package
 from patchman.signals import pbar_start, pbar_update
-from util import extract, fetch_content, get_setting_of_type, get_url
-from util.logging import error_message, warning_message
+from util import (
+    extract, fetch_content, get_setting_of_type, get_url, run_concurrently,
+)
+from util.logging import clear_forked_pbar, error_message, warning_message
 
 DSCs = {}
 
 
-def update_debian_errata(concurrent_processing=True):
+def update_debian_errata(concurrent_processing=True, max_workers=25):
     """ Update Debian errata using:
           https://salsa.debian.org/security-tracker-team/security-tracker/raw/master/data/DSA/list
           https://salsa.debian.org/security-tracker-team/security-tracker/raw/master/data/DSA/list
@@ -47,7 +48,7 @@ def update_debian_errata(concurrent_processing=True):
     fetch_dscs_from_debian_package_file_maps()
     accepted_codenames = get_accepted_debian_codenames()
     errata = parse_debian_errata(advisories, accepted_codenames)
-    create_debian_errata(errata, accepted_codenames, concurrent_processing)
+    create_debian_errata(errata, accepted_codenames, concurrent_processing, max_workers)
 
 
 def fetch_debian_dsa_advisories():
@@ -113,7 +114,9 @@ def parse_debian_errata(advisories, accepted_codenames):
     """
     distro_pattern = re.compile(r'^\t\[(.+?)\] - .*')
     title_pattern = re.compile(r'^\[(.+?)\] (.+?) (.+?)[ ]+[-]+ (.*)')
+    distro_package_pattern = re.compile(r'^\t\[(.+?)\] - (.+?) (.*)')
     errata = []
+    dsc_fetches = []
     e = {'packages': {}, 'cve_ids': [], 'releases': []}
     for line in advisories.splitlines():
         if line.startswith('['):
@@ -132,9 +135,17 @@ def parse_debian_errata(advisories, accepted_codenames):
                 e['releases'].append(release)
                 if not e.get('packages').get(release):
                     e['packages'][release] = []
-                e['packages'][release].append(parse_debian_erratum_package(line, accepted_codenames))
+                pkg_match = re.match(distro_package_pattern, line)
+                if pkg_match and pkg_match.group(1) in accepted_codenames:
+                    source_package = pkg_match.group(2)
+                    source_version = pkg_match.group(3)
+                    dsc_fetches.append((source_package, source_version))
+                    e['packages'][release].append((source_package, source_version))
+                else:
+                    e['packages'][release].append(None)
     # add the last one
     errata = add_errata_by_codename(errata, e, accepted_codenames)
+    fetch_debian_dsc_package_lists(dsc_fetches)
     return errata
 
 
@@ -162,11 +173,11 @@ def parse_debian_erratum_advisory(e, match):
     return e
 
 
-def create_debian_errata(errata, accepted_codenames, concurrent_processing):
+def create_debian_errata(errata, accepted_codenames, concurrent_processing, max_workers=25):
     """ Create Debian Errata
     """
     if concurrent_processing:
-        create_debian_errata_concurrently(errata, accepted_codenames)
+        create_debian_errata_concurrently(errata, accepted_codenames, max_workers)
     else:
         create_debian_errata_serially(errata, accepted_codenames)
 
@@ -181,25 +192,25 @@ def create_debian_errata_serially(errata, accepted_codenames):
         pbar_update.send(sender=None, index=i + 1)
 
 
-def create_debian_errata_concurrently(errata, accepted_codenames):
+def create_debian_errata_concurrently(errata, accepted_codenames, max_workers=25):
     """ Create Debian Errata concurrently
     """
-    connections.close_all()
     elen = len(errata)
     pbar_start.send(sender=None, ptext=f'Processing {elen} Debian Errata', plen=elen)
-    i = 0
-    with concurrent.futures.ProcessPoolExecutor(max_workers=25) as executor:
-        futures = [executor.submit(process_debian_erratum, erratum, accepted_codenames) for erratum in errata]
-        for future in concurrent.futures.as_completed(futures):
-            i += 1
-            pbar_update.send(sender=None, index=i + 1)
+    args = [(erratum, accepted_codenames) for erratum in errata]
+    for i, _ in enumerate(run_concurrently(_process_debian_erratum_wrapper, args, max_workers)):
+        pbar_update.send(sender=None, index=i + 1)
+
+
+def _process_debian_erratum_wrapper(args):
+    clear_forked_pbar()
+    return process_debian_erratum(*args)
 
 
 def process_debian_erratum(erratum, accepted_codenames):
     """ Process a single Debian Erratum
     """
     try:
-        from errata.utils import get_or_create_erratum
         erratum_name = erratum.get('name')
         e, created = get_or_create_erratum(
             name=erratum_name,
@@ -221,19 +232,13 @@ def process_debian_erratum(erratum, accepted_codenames):
         error_message(text=exc)
 
 
-def parse_debian_erratum_package(line, accepted_codenames):
-    """ Parse the codename and source package from a DSA/DLA file
-        Returns the source package and source version
+def fetch_debian_dsc_package_lists(dsc_fetches):
+    """ Fetch DSC package lists with a progress bar
     """
-    distro_package_pattern = re.compile(r'^\t\[(.+?)\] - (.+?) (.*)')
-    match = re.match(distro_package_pattern, line)
-    if match:
-        codename = match.group(1)
-        if codename in accepted_codenames:
-            source_package = match.group(2)
-            source_version = match.group(3)
-            fetch_debian_dsc_package_list(source_package, source_version)
-            return source_package, source_version
+    pbar_start.send(sender=None, ptext=f'Fetching {len(dsc_fetches)} Debian DSC files', plen=len(dsc_fetches))
+    for i, (package, version) in enumerate(dsc_fetches):
+        fetch_debian_dsc_package_list(package, version)
+        pbar_update.send(sender=None, index=i + 1)
 
 
 def get_debian_dsc_package_list(package, version):
@@ -301,6 +306,8 @@ def create_debian_os_releases(codename_to_version):
 def process_debian_erratum_fixed_packages(e, package_data):
     """ Process packages fixed in a Debian errata
     """
+    if not package_data:
+        return
     source_package, source_version = package_data
     epoch, ver, rel = find_evr(source_version)
     package_list = get_debian_dsc_package_list(source_package, source_version)