Merge pull request #8952 from readthedocs/humitos/celery-upgrade

Author: humitos

readthedocs/builds/apps.py

@@ -12,20 +12,4 @@ class Config(AppConfig):
     verbose_name = _("Builds")
 
     def ready(self):
-        from readthedocs.builds.tasks import ArchiveBuilds
-        from readthedocs.worker import app
-        app.tasks.register(ArchiveBuilds)
-
-        try:
-            from readthedocsext.monitoring.metrics.tasks import (
-                Metrics1mTask,
-                Metrics5mTask,
-                Metrics10mTask,
-                Metrics30mTask,
-            )
-            app.tasks.register(Metrics1mTask)
-            app.tasks.register(Metrics5mTask)
-            app.tasks.register(Metrics10mTask)
-            app.tasks.register(Metrics30mTask)
-        except (ModuleNotFoundError, ImportError):
-            log.warning('Metrics tasks could not be imported.')
+        import readthedocs.builds.tasks

readthedocs/builds/tasks.py

@@ -178,67 +178,61 @@ def _get_version(self, task, args, kwargs):
         return version
 
 
-class ArchiveBuilds(Task):
-
-    """Task to archive old builds to cold storage."""
-
-    name = __name__ + '.archive_builds'
-
-    def run(self, *args, **kwargs):
-        if not settings.RTD_SAVE_BUILD_COMMANDS_TO_STORAGE:
-            return
-
-        lock_id = '{0}-lock'.format(self.name)
-        days = kwargs.get('days', 14)
-        limit = kwargs.get('limit', 2000)
-        delete = kwargs.get('delete', True)
-
-        with memcache_lock(lock_id, self.app.oid) as acquired:
-            if acquired:
-                archive_builds_task(days=days, limit=limit, delete=delete)
-            else:
-                log.warning('Archive Builds Task still locked')
+# NOTE: re-define this task to keep the old name. Delete it after deploy.
+@app.task(queue='web')
+def archive_builds(days=14, limit=200, include_cold=False, delete=False):
+    archive_builds_task.delay(days=days, limit=limit, include_cold=include_cold, delete=delete)
 
 
-def archive_builds_task(days=14, limit=200, include_cold=False, delete=False):
+@app.task(queue='web', bind=True)
+def archive_builds_task(self, days=14, limit=200, include_cold=False, delete=False):
     """
-    Find stale builds and remove build paths.
+    Task to archive old builds to cold storage.
 
     :arg days: Find builds older than `days` days.
     :arg include_cold: If True, include builds that are already in cold storage
     :arg delete: If True, deletes BuildCommand objects after archiving them
     """
-    max_date = timezone.now() - timezone.timedelta(days=days)
-    queryset = Build.objects.exclude(commands__isnull=True)
-    if not include_cold:
-        queryset = queryset.exclude(cold_storage=True)
-
-    queryset = (
-        queryset
-        .filter(date__lt=max_date)
-        .prefetch_related('commands')
-        .only('date', 'cold_storage')
-        [:limit]
-    )
+    if not settings.RTD_SAVE_BUILD_COMMANDS_TO_STORAGE:
+        return
 
-    for build in queryset:
-        commands = BuildCommandSerializer(build.commands, many=True).data
-        if commands:
-            for cmd in commands:
-                if len(cmd['output']) > MAX_BUILD_COMMAND_SIZE:
-                    cmd['output'] = cmd['output'][-MAX_BUILD_COMMAND_SIZE:]
-                    cmd['output'] = "... (truncated) ...\n\nCommand output too long. Truncated to last 1MB.\n\n" + cmd['output']  # noqa
-                    log.warning('Truncating build command for build.', build_id=build.id)
-            output = BytesIO(json.dumps(commands).encode('utf8'))
-            filename = '{date}/{id}.json'.format(date=str(build.date.date()), id=build.id)
-            try:
-                build_commands_storage.save(name=filename, content=output)
-                build.cold_storage = True
-                build.save()
-                if delete:
-                    build.commands.all().delete()
-            except IOError:
-                log.exception('Cold Storage save failure')
+    lock_id = '{0}-lock'.format(self.name)
+    with memcache_lock(lock_id, self.app.oid) as acquired:
+        if not acquired:
+            log.warning('Archive Builds Task still locked')
+            return False
+
+        max_date = timezone.now() - timezone.timedelta(days=days)
+        queryset = Build.objects.exclude(commands__isnull=True)
+        if not include_cold:
+            queryset = queryset.exclude(cold_storage=True)
+
+        queryset = (
+            queryset
+            .filter(date__lt=max_date)
+            .prefetch_related('commands')
+            .only('date', 'cold_storage')
+            [:limit]
+        )
+
+        for build in queryset:
+            commands = BuildCommandSerializer(build.commands, many=True).data
+            if commands:
+                for cmd in commands:
+                    if len(cmd['output']) > MAX_BUILD_COMMAND_SIZE:
+                        cmd['output'] = cmd['output'][-MAX_BUILD_COMMAND_SIZE:]
+                        cmd['output'] = "... (truncated) ...\n\nCommand output too long. Truncated to last 1MB.\n\n" + cmd['output']  # noqa
+                        log.warning('Truncating build command for build.', build_id=build.id)
+                output = BytesIO(json.dumps(commands).encode('utf8'))
+                filename = '{date}/{id}.json'.format(date=str(build.date.date()), id=build.id)
+                try:
+                    build_commands_storage.save(name=filename, content=output)
+                    build.cold_storage = True
+                    build.save()
+                    if delete:
+                        build.commands.all().delete()
+                except IOError:
+                    log.exception('Cold Storage save failure')
 
 
 @app.task(queue='web')

readthedocs/builds/tests/test_tasks.py

@@ -1,7 +1,7 @@
 from datetime import datetime, timedelta
 from unittest import mock
 
-from django.test import TestCase
+from django.test import TestCase, override_settings
 from django.utils import timezone
 from django_dynamic_fixture import get
 
@@ -74,6 +74,7 @@ def test_delete_inactive_external_versions(self):
         self.assertEqual(Version.external.all().count(), 2)
         self.assertFalse(Version.objects.filter(slug='external-inactive-old').exists())
 
+    @override_settings(RTD_SAVE_BUILD_COMMANDS_TO_STORAGE=True)
     @mock.patch('readthedocs.builds.tasks.build_commands_storage')
     def test_archive_builds(self, build_commands_storage):
         project = get(Project)
@@ -98,7 +99,7 @@ def test_archive_builds(self, build_commands_storage):
         self.assertEqual(Build.objects.count(), 10)
         self.assertEqual(BuildCommandResult.objects.count(), 100)
 
-        archive_builds_task(days=5, delete=True)
+        archive_builds_task.delay(days=5, delete=True)
 
         self.assertEqual(len(build_commands_storage.save.mock_calls), 5)
         self.assertEqual(Build.objects.count(), 10)

readthedocs/builds/utils.py

@@ -94,8 +94,9 @@ def memcache_lock(lock_id, oid):
     finally:
         # memcache delete is very slow, but we have to use it to take
         # advantage of using add() for atomic locking
-        if monotonic() < timeout_at:
+        if monotonic() < timeout_at and status:
             # don't release the lock if we exceeded the timeout
             # to lessen the chance of releasing an expired lock
-            # owned by someone else.
+            # owned by someone else
+            # also don't release the lock if we didn't acquire it
             cache.delete(lock_id)

readthedocs/core/apps.py

@@ -1,9 +1,11 @@
-# -*- coding: utf-8 -*-
-
 """App configurations for core app."""
 
+import structlog
+
 from django.apps import AppConfig
 
+log = structlog.get_logger(__name__)
+
 
 class CoreAppConfig(AppConfig):
     name = 'readthedocs.core'
@@ -14,3 +16,8 @@ def ready(self):
 
         # Import `readthedocs.core.logs` to set up structlog
         import readthedocs.core.logs  # noqa
+
+        try:
+            import readthedocsext.monitoring.metrics.tasks
+        except (ModuleNotFoundError, ImportError):
+            log.info('Metrics tasks could not be imported.')

readthedocs/projects/tasks/builds.py

@@ -205,9 +205,6 @@ def update_versions_from_repository(self, environment):
     bind=True,
 )
 def sync_repository_task(self, version_id):
-    # NOTE: `before_start` is new on Celery 5.2.x, but we are using 5.1.x currently.
-    self.before_start(self.request.id, self.request.args, self.request.kwargs)
-
     self.execute()
 
 
@@ -993,7 +990,4 @@ def is_type_sphinx(self):
     bind=True,
 )
 def update_docs_task(self, version_id, build_id, build_commit=None):
-    # NOTE: `before_start` is new on Celery 5.2.x, but we are using 5.1.x currently.
-    self.before_start(self.request.id, self.request.args, self.request.kwargs)
-
     self.execute()

readthedocs/settings/base.py

@@ -432,6 +432,8 @@ def TEMPLATES(self):
             'options': {'queue': 'web'},
             'kwargs': {
                 'days': 1,
+                'limit': 2000,
+                'delete': True,
             },
         },
         'every-day-delete-inactive-external-versions': {

requirements/pip.txt

@@ -38,13 +38,7 @@ Pygments==2.11.2
 # django-redis-cache 2.1.3 depends on redis<4.0
 redis==3.5.3  # pyup: ignore
 
-# Pin celery and kombu because it produces an error
-# File "/home/docs/lib/python3.8/site-packages/celery/app/trace.py", line 361, in build_tracer
-#     push_request = request_stack.push
-# AttributeError: 'NoneType' object has no attribute 'push'
-# See https://github.com/celery/celery/discussions/7172
-kombu==5.1.0  # pyup: ignore
-celery==5.1.2  # pyup: ignore
+celery==5.2.3
 
 # When upgrading to 0.43.0 we should double check the ``base.html`` change
 # described in the changelog. In previous versions, the allauth app included a