Use new Celery, use new application pattern

* Use modern celery
* Drop djcelery
* New pattern for starting celery
* Bump redis to 2.10.6 to avoid startup bug, change autodiscover call

Author: agjohnson

readthedocs/__init__.py

@@ -0,0 +1 @@
+from .tasks import app

readthedocs/core/management/commands/update_api.py

@@ -33,4 +33,5 @@ def handle(self, *args, **options):
             project_data = api.project(slug).get()
             p = APIProject(**project_data)
             log.info("Building %s", p)
-            tasks.update_docs.run(pk=p.pk, docker=docker)
+            update_docs = tasks.UpdateDocsTask()
+            update_docs.run(pk=p.pk, docker=docker)

readthedocs/core/management/commands/update_repos.py

@@ -54,9 +54,11 @@ def handle(self, *args, **options):
                     for version in Version.objects.filter(project__slug=slug,
                                                           active=True,
                                                           uploaded=False):
-                        tasks.update_docs.run(pk=version.project_id,
-                                              record=False,
-                                              version_pk=version.pk)
+                        tasks.UpdateDocsTask().run(
+                            pk=version.project_id,
+                            record=False,
+                            version_pk=version.pk
+                        )
                 else:
                     p = Project.all_objects.get(slug=slug)
                     log.info("Building %s", p)
@@ -66,12 +68,17 @@ def handle(self, *args, **options):
                 log.info("Updating all versions")
                 for version in Version.objects.filter(active=True,
                                                       uploaded=False):
-                    tasks.update_docs.run(pk=version.project_id,
-                                          record=record,
-                                          force=force,
-                                          version_pk=version.pk)
+                    tasks.UpdateDocsTask().run(
+                        pk=version.project_id,
+                        record=record,
+                        force=force,
+                        version_pk=version.pk
+                    )
             else:
                 log.info("Updating all docs")
                 for project in Project.objects.all():
-                    tasks.update_docs.run(pk=project.pk, record=record,
-                                          force=force)
+                    tasks.UpdateDocsTask().run(
+                        pk=project.pk,
+                        record=record,
+                        force=force
+                    )

readthedocs/core/management/commands/update_versions.py

@@ -4,7 +4,7 @@
 from django.core.management.base import BaseCommand
 
 from readthedocs.builds.models import Version
-from readthedocs.projects.tasks import update_docs
+from readthedocs.projects.tasks import UpdateDocsTask
 
 
 class Command(BaseCommand):
@@ -13,5 +13,6 @@ class Command(BaseCommand):
 
     def handle(self, *args, **options):
         for version in Version.objects.filter(active=True, built=False):
+            update_docs = UpdateDocsTask()
             update_docs.run(version.project_id, record=False,
                             version_pk=version.pk)

readthedocs/core/tasks.py

@@ -3,7 +3,7 @@
 from __future__ import absolute_import
 import logging
 
-from celery import task
+from celery import shared_task
 from django.conf import settings
 from django.core.mail import EmailMultiAlternatives
 from django.template.loader import get_template
@@ -15,7 +15,7 @@
 EMAIL_TIME_LIMIT = 30
 
 
-@task(queue='web', time_limit=EMAIL_TIME_LIMIT)
+@shared_task(queue='web', time_limit=EMAIL_TIME_LIMIT)
 def send_email_task(recipient, subject, template, template_html, context=None):
     """Send multipart email
 

readthedocs/core/utils/__init__.py

@@ -77,7 +77,7 @@ def trigger_build(project, version=None, record=True, force=False, basic=False):
     will be prefixed with ``build-`` to unify build queue names.
     """
     # Avoid circular import
-    from readthedocs.projects.tasks import update_docs
+    from readthedocs.projects.tasks import UpdateDocsTask
     from readthedocs.builds.models import Build
 
     if project.skip:
@@ -121,6 +121,7 @@ def trigger_build(project, version=None, record=True, force=False, basic=False):
     options['soft_time_limit'] = time_limit
     options['time_limit'] = int(time_limit * 1.2)
 
+    update_docs = UpdateDocsTask()
     update_docs.apply_async(kwargs=kwargs, **options)
 
     return build

readthedocs/core/utils/tasks/retrieve.py

@@ -1,7 +1,6 @@
 """Utilities for retrieving task data."""
 
 from __future__ import absolute_import
-from djcelery import celery as celery_app
 from celery.result import AsyncResult
 
 
@@ -20,14 +19,16 @@ def get_task_data(task_id):
 
     meta data has no ``'task_name'`` key set.
     """
+    from readthedocs.worker import app
+
     result = AsyncResult(task_id)
     state, info = result.state, result.info
     if state == 'PENDING':
         raise TaskNotFound(task_id)
     if 'task_name' not in info:
         raise TaskNotFound(task_id)
     try:
-        task = celery_app.tasks[info['task_name']]
+        task = app.tasks[info['task_name']]
     except KeyError:
         raise TaskNotFound(task_id)
     return task, state, info

readthedocs/oauth/tasks.py

@@ -2,7 +2,6 @@
 
 from __future__ import absolute_import
 from django.contrib.auth.models import User
-from djcelery import celery as celery_app
 
 from readthedocs.core.utils.tasks import PublicTask
 from readthedocs.core.utils.tasks import permission_check
@@ -22,4 +21,4 @@ def run_public(self, user_id):
                 service.sync()
 
 
-sync_remote_repositories = celery_app.tasks[SyncRemoteRepositories.name]
+sync_remote_repositories = SyncRemoteRepositories()

readthedocs/projects/tasks.py

@@ -16,7 +16,7 @@
 
 import requests
 from builtins import str
-from celery import task, Task
+from celery import shared_task, Task
 from celery.exceptions import SoftTimeLimitExceeded
 from django.conf import settings
 from django.core.urlresolvers import reverse
@@ -53,6 +53,7 @@
 from readthedocs.search.parse_json import process_all_json_files
 from readthedocs.search.utils import process_mkdocs_json
 from readthedocs.vcs_support import utils as vcs_support_utils
+from readthedocs.worker import app
 
 
 log = logging.getLogger(__name__)
@@ -80,7 +81,7 @@ class UpdateDocsTask(Task):
 
     max_retries = 5
     default_retry_delay = (7 * 60)
-    name = 'update_docs'
+    name = __name__ + '.update_docs'
 
     def __init__(self, build_env=None, python_env=None, config=None,
                  force=False, search=True, localmedia=True,
@@ -490,10 +491,10 @@ def send_notifications(self):
         send_notifications.delay(self.version.pk, build_pk=self.build['id'])
 
 
-update_docs = celery_app.tasks[UpdateDocsTask.name]
+app.register_task(UpdateDocsTask())
 
 
-@task()
+@shared_task()
 def update_imported_docs(version_pk):
     """
     Check out or update the given project's repository
@@ -572,7 +573,7 @@ def update_imported_docs(version_pk):
 
 
 # Web tasks
-@task(queue='web')
+@shared_task(queue='web')
 def sync_files(project_pk, version_pk, hostname=None, html=False,
                localmedia=False, search=False, pdf=False, epub=False):
     """Sync build artifacts to application instances
@@ -604,7 +605,7 @@ def sync_files(project_pk, version_pk, hostname=None, html=False,
     update_static_metadata(project_pk)
 
 
-@task(queue='web')
+@shared_task(queue='web')
 def move_files(version_pk, hostname, html=False, localmedia=False, search=False,
                pdf=False, epub=False):
     """Task to move built documentation to web servers
@@ -670,7 +671,7 @@ def move_files(version_pk, hostname, html=False, localmedia=False, search=False,
             Syncer.copy(from_path, to_path, host=hostname)
 
 
-@task(queue='web')
+@shared_task(queue='web')
 def update_search(version_pk, commit, delete_non_commit_files=True):
     """Task to update search indexes
 
@@ -705,15 +706,15 @@ def update_search(version_pk, commit, delete_non_commit_files=True):
     )
 
 
-@task(queue='web')
+@shared_task(queue='web')
 def symlink_project(project_pk):
     project = Project.objects.get(pk=project_pk)
     for symlink in [PublicSymlink, PrivateSymlink]:
         sym = symlink(project=project)
         sym.run()
 
 
-@task(queue='web')
+@shared_task(queue='web')
 def symlink_domain(project_pk, domain_pk, delete=False):
     project = Project.objects.get(pk=project_pk)
     domain = Domain.objects.get(pk=domain_pk)
@@ -725,15 +726,15 @@ def symlink_domain(project_pk, domain_pk, delete=False):
             sym.symlink_cnames(domain)
 
 
-@task(queue='web')
+@shared_task(queue='web')
 def symlink_subproject(project_pk):
     project = Project.objects.get(pk=project_pk)
     for symlink in [PublicSymlink, PrivateSymlink]:
         sym = symlink(project=project)
         sym.symlink_subprojects()
 
 
-@task(queue='web')
+@shared_task(queue='web')
 def fileify(version_pk, commit):
     """
     Create ImportedFile objects for all of a version's files.
@@ -809,7 +810,7 @@ def _manage_imported_files(version, path, commit):
             purge(cdn_ids[version.project.slug], changed_files)
 
 
-@task(queue='web')
+@shared_task(queue='web')
 def send_notifications(version_pk, build_pk):
     version = Version.objects.get(pk=version_pk)
     build = Build.objects.get(pk=build_pk)
@@ -878,7 +879,7 @@ def webhook_notification(version, build, hook_url):
     requests.post(hook_url, data=data)
 
 
-@task(queue='web')
+@shared_task(queue='web')
 def update_static_metadata(project_pk, path=None):
     """Update static metadata JSON file
 
@@ -924,7 +925,7 @@ def update_static_metadata(project_pk, path=None):
 
 
 # Random Tasks
-@task()
+@shared_task()
 def remove_dir(path):
     """
     Remove a directory on the build/celery server.
@@ -936,7 +937,7 @@ def remove_dir(path):
     shutil.rmtree(path, ignore_errors=True)
 
 
-@task()
+@shared_task()
 def clear_artifacts(version_pk):
     """Remove artifacts from the web servers"""
     version = Version.objects.get(pk=version_pk)
@@ -946,31 +947,31 @@ def clear_artifacts(version_pk):
     clear_html_artifacts(version)
 
 
-@task()
+@shared_task()
 def clear_pdf_artifacts(version):
     if isinstance(version, int):
         version = Version.objects.get(pk=version)
     remove_dir(version.project.get_production_media_path(
         type_='pdf', version_slug=version.slug))
 
 
-@task()
+@shared_task()
 def clear_epub_artifacts(version):
     if isinstance(version, int):
         version = Version.objects.get(pk=version)
     remove_dir(version.project.get_production_media_path(
         type_='epub', version_slug=version.slug))
 
 
-@task()
+@shared_task()
 def clear_htmlzip_artifacts(version):
     if isinstance(version, int):
         version = Version.objects.get(pk=version)
     remove_dir(version.project.get_production_media_path(
         type_='htmlzip', version_slug=version.slug))
 
 
-@task()
+@shared_task()
 def clear_html_artifacts(version):
     if isinstance(version, int):
         version = Version.objects.get(pk=version)

readthedocs/rtd_tests/tests/test_celery.py

@@ -72,7 +72,8 @@ def test_update_docs(self):
         build = get(Build, project=self.project,
                     version=self.project.versions.first())
         with mock_api(self.repo) as mapi:
-            result = tasks.update_docs.delay(
+            update_docs = tasks.UpdateDocsTask()
+            result = update_docs.delay(
                 self.project.pk,
                 build_pk=build.pk,
                 record=False,
@@ -89,7 +90,8 @@ def test_update_docs_unexpected_setup_exception(self, mock_update_build, mock_se
         build = get(Build, project=self.project,
                     version=self.project.versions.first())
         with mock_api(self.repo) as mapi:
-            result = tasks.update_docs.delay(
+            update_docs = tasks.UpdateDocsTask()
+            result = update_docs.delay(
                 self.project.pk,
                 build_pk=build.pk,
                 record=False,
@@ -107,7 +109,8 @@ def test_update_docs_unexpected_build_exception(self, mock_update_build, mock_bu
         build = get(Build, project=self.project,
                     version=self.project.versions.first())
         with mock_api(self.repo) as mapi:
-            result = tasks.update_docs.delay(
+            update_docs = tasks.UpdateDocsTask()
+            result = update_docs.delay(
                 self.project.pk,
                 build_pk=build.pk,
                 record=False,

readthedocs/rtd_tests/tests/test_core_utils.py

@@ -18,12 +18,12 @@ def setUp(self):
         self.project = get(Project, container_time_limit=None)
         self.version = get(Version, project=self.project)
 
-    @mock.patch('readthedocs.projects.tasks.update_docs')
+    @mock.patch('readthedocs.projects.tasks.UpdateDocsTask')
     def test_trigger_build_time_limit(self, update_docs):
         """Pass of time limit"""
         trigger_build(project=self.project, version=self.version)
-        update_docs.assert_has_calls([
-            mock.call.apply_async(
+        update_docs().apply_async.assert_has_calls([
+            mock.call(
                 time_limit=720,
                 soft_time_limit=600,
                 queue=mock.ANY,
@@ -38,13 +38,13 @@ def test_trigger_build_time_limit(self, update_docs):
             )
         ])
 
-    @mock.patch('readthedocs.projects.tasks.update_docs')
+    @mock.patch('readthedocs.projects.tasks.UpdateDocsTask')
     def test_trigger_build_invalid_time_limit(self, update_docs):
         """Time limit as string"""
         self.project.container_time_limit = '200s'
         trigger_build(project=self.project, version=self.version)
-        update_docs.assert_has_calls([
-            mock.call.apply_async(
+        update_docs().apply_async.assert_has_calls([
+            mock.call(
                 time_limit=720,
                 soft_time_limit=600,
                 queue=mock.ANY,
@@ -59,13 +59,13 @@ def test_trigger_build_invalid_time_limit(self, update_docs):
             )
         ])
 
-    @mock.patch('readthedocs.projects.tasks.update_docs')
+    @mock.patch('readthedocs.projects.tasks.UpdateDocsTask')
     def test_trigger_build_rounded_time_limit(self, update_docs):
         """Time limit should round down"""
         self.project.container_time_limit = 3
         trigger_build(project=self.project, version=self.version)
-        update_docs.assert_has_calls([
-            mock.call.apply_async(
+        update_docs().apply_async.assert_has_calls([
+            mock.call(
                 time_limit=3,
                 soft_time_limit=3,
                 queue=mock.ANY,