use redis for caching and use locks for tasks

Author: furlongm

errata/tasks.py

@@ -16,13 +16,15 @@
 
 from celery import shared_task
 
+from django.core.cache import cache
+
 from errata.sources.distros.arch import update_arch_errata
 from errata.sources.distros.alma import update_alma_errata
 from errata.sources.distros.debian import update_debian_errata
 from errata.sources.distros.centos import update_centos_errata
 from errata.sources.distros.rocky import update_rocky_errata
 from errata.sources.distros.ubuntu import update_ubuntu_errata
-from util.logging import error_message
+from util.logging import error_message, warning_message
 from repos.models import Repository
 from security.tasks import update_cves, update_cwes
 from util import get_setting_of_type
@@ -44,34 +46,44 @@ def update_yum_repo_errata(repo_id=None, force=False):
 def update_errata(erratum_type=None, force=False, repo=None):
     """ Update all distros errata
     """
-    errata_os_updates = []
-    erratum_types = ['yum', 'rocky', 'alma', 'arch', 'ubuntu', 'debian', 'centos']
-    erratum_type_defaults = ['yum', 'rocky', 'alma', 'arch', 'ubuntu', 'debian']
-    if erratum_type:
-        if erratum_type not in erratum_types:
-            error_message.send(sender=None, text=f'Erratum type `{erratum_type}` not in {erratum_types}')
-        else:
-            errata_os_updates = erratum_type
+    lock_key = 'update_errata_lock'
+    # lock will expire after 48 hours
+    lock_expire = 60 * 60 * 48
+
+    if cache.add(lock_key, 'true', lock_expire):
+        try:
+            errata_os_updates = []
+            erratum_types = ['yum', 'rocky', 'alma', 'arch', 'ubuntu', 'debian', 'centos']
+            erratum_type_defaults = ['yum', 'rocky', 'alma', 'arch', 'ubuntu', 'debian']
+            if erratum_type:
+                if erratum_type not in erratum_types:
+                    error_message(text=f'Erratum type `{erratum_type}` not in {erratum_types}')
+                else:
+                    errata_os_updates = erratum_type
+            else:
+                errata_os_updates = get_setting_of_type(
+                    setting_name='ERRATA_OS_UPDATES',
+                    setting_type=list,
+                    default=erratum_type_defaults,
+                )
+            if 'yum' in errata_os_updates:
+                update_yum_repo_errata(repo_id=repo, force=force)
+            if 'arch' in errata_os_updates:
+                update_arch_errata()
+            if 'alma' in errata_os_updates:
+                update_alma_errata()
+            if 'rocky' in errata_os_updates:
+                update_rocky_errata()
+            if 'debian' in errata_os_updates:
+                update_debian_errata()
+            if 'ubuntu' in errata_os_updates:
+                update_ubuntu_errata()
+            if 'centos' in errata_os_updates:
+                update_centos_errata()
+        finally:
+            cache.delete(lock_key)
     else:
-        errata_os_updates = get_setting_of_type(
-            setting_name='ERRATA_OS_UPDATES',
-            setting_type=list,
-            default=erratum_type_defaults,
-        )
-    if 'yum' in errata_os_updates:
-        update_yum_repo_errata(repo_id=repo, force=force)
-    if 'arch' in errata_os_updates:
-        update_arch_errata()
-    if 'alma' in errata_os_updates:
-        update_alma_errata()
-    if 'rocky' in errata_os_updates:
-        update_rocky_errata()
-    if 'debian' in errata_os_updates:
-        update_debian_errata()
-    if 'ubuntu' in errata_os_updates:
-        update_ubuntu_errata()
-    if 'centos' in errata_os_updates:
-        update_centos_errata()
+        warning_message('Already updating Errata, skipping task.')
 
 
 @shared_task

etc/patchman/local_settings.py

@@ -44,22 +44,16 @@
 # Whether to run patchman under the gunicorn web server
 RUN_GUNICORN = False
 
+# Set the default timeout to e.g. 30 seconds to enable UI caching
+# Note that the UI results may be out of date for this amount of time
 CACHES = {
     'default': {
-        'BACKEND': 'django.core.cache.backends.dummy.DummyCache',
+        'BACKEND': 'django.core.cache.backends.redis.RedisCache',
+        'LOCATION': 'redis://127.0.0.1:6379',
+        'TIMEOUT': 0,
     }
 }
 
-# Uncomment to enable redis caching for e.g. 30 seconds
-# Note that the UI results may be out of date for this amount of time
-# CACHES = {
-#     'default': {
-#         'BACKEND': 'django.core.cache.backends.redis.RedisCache',
-#         'LOCATION': 'redis://127.0.0.1:6379',
-#         'TIMEOUT': 30,
-#     }
-# }
-
 from datetime import timedelta        # noqa
 from celery.schedules import crontab  # noqa
 CELERY_BEAT_SCHEDULE = {

reports/tasks.py

@@ -17,19 +17,30 @@
 
 from celery import shared_task
 
+from django.core.cache import cache
 from django.db.utils import OperationalError
 
 from hosts.models import Host
 from reports.models import Report
-from util.logging import info_message
+from util.logging import info_message, warning_message
 
 
 @shared_task(bind=True, autoretry_for=(OperationalError,), retry_backoff=True, retry_kwargs={'max_retries': 5})
 def process_report(self, report_id):
     """ Task to process a single report
     """
     report = Report.objects.get(id=report_id)
-    report.process()
+    lock_key = f'process_report_lock_{report_id}'
+    # lock will expire after 1 hour
+    lock_expire = 60 * 60
+
+    if cache.add(lock_key, 'true', lock_expire):
+        try:
+            report.process()
+        finally:
+            cache.delete(lock_key)
+    else:
+        warning_message(f'Already processing report {report_id}, skipping task.')
 
 
 @shared_task

repos/tasks.py

@@ -16,7 +16,10 @@
 
 from celery import shared_task
 
+from django.core.cache import cache
+
 from repos.models import Repository
+from util.logging import warning_message
 
 
 @shared_task
@@ -32,5 +35,15 @@ def refresh_repos(force=False):
     """ Refresh metadata for all enabled repos
     """
     repos = Repository.objects.filter(enabled=True)
-    for repo in repos:
-        refresh_repo.delay(repo.id, force)
+    lock_key = 'refresh_repos_lock'
+    # lock will expire after 1 day
+    lock_expire = 60 * 60 * 24
+
+    if cache.add(lock_key, 'true', lock_expire):
+        try:
+            for repo in repos:
+                refresh_repo.delay(repo.id, force)
+        finally:
+            cache.delete(lock_key)
+    else:
+        warning_message('Already refreshing repos, skipping task.')

security/tasks.py

@@ -16,7 +16,10 @@
 
 from celery import shared_task
 
+from django.core.cache import cache
+
 from security.models import CVE, CWE
+from util.logging import warning_message
 
 
 @shared_task
@@ -31,8 +34,18 @@ def update_cve(cve_id):
 def update_cves():
     """ Task to update all CVEs
     """
-    for cve in CVE.objects.all():
-        update_cve.delay(cve.id)
+    lock_key = 'update_cves_lock'
+    # lock will expire after 1 week
+    lock_expire = 60 * 60 * 168
+
+    if cache.add(lock_key, 'true', lock_expire):
+        try:
+            for cve in CVE.objects.all():
+                update_cve.delay(cve.id)
+        finally:
+            cache.delete(lock_key)
+    else:
+        warning_message('Already updating CVEs, skipping task.')
 
 
 @shared_task
@@ -45,7 +58,17 @@ def update_cwe(cwe_id):
 
 @shared_task
 def update_cwes():
-    """ Task to update all CWEa
+    """ Task to update all CWEs
     """
-    for cwe in CWE.objects.all():
-        update_cwe.delay(cwe.id)
+    lock_key = 'update_cwes_lock'
+    # lock will expire after 1 week
+    lock_expire = 60 * 60 * 168
+
+    if cache.add(lock_key, 'true', lock_expire):
+        try:
+            for cwe in CWE.objects.all():
+                update_cwe.delay(cwe.id)
+        finally:
+            cache.delete(lock_key)
+    else:
+        warning_message('Already updating CWEs, skipping task.')