Merge pull request #2946 from rtfd/broadcast-fixes

Move move_files to a broadcast model.

Author: ericholscher

readthedocs/builds/models.py

@@ -163,7 +163,7 @@ def save(self, *args, **kwargs):  # pylint: disable=arguments-differ
     def delete(self, *args, **kwargs):  # pylint: disable=arguments-differ
         from readthedocs.projects import tasks
         log.info('Removing files for version %s', self.slug)
-        tasks.clear_artifacts.delay(version_pk=self.pk)
+        broadcast(type='app', task=tasks.clear_artifacts, args=[self.pk])
         broadcast(type='app', task=tasks.symlink_project, args=[self.project.pk])
         super(Version, self).delete(*args, **kwargs)
 

readthedocs/core/management/commands/set_metadata.py

@@ -7,6 +7,7 @@
 
 from readthedocs.projects import tasks
 from readthedocs.projects.models import Project
+from readthedocs.core.utils import broadcast
 
 log = logging.getLogger(__name__)
 
@@ -20,6 +21,6 @@ def handle(self, *args, **options):
         for p in queryset:
             log.info("Generating metadata for %s", p)
             try:
-                tasks.update_static_metadata(p.pk)
+                broadcast(type='app', task=tasks.update_static_metadata, args=[p.pk])
             except Exception:
                 log.error('Build failed for %s', p, exc_info=True)

readthedocs/core/utils/__init__.py

@@ -28,22 +28,10 @@
 SYNC_USER = getattr(settings, 'SYNC_USER', getpass.getuser())
 
 
-def run_on_app_servers(command):
-    """A helper to copy a single file across app servers"""
-    log.info("Running %s on app servers", command)
-    ret_val = 0
-    if getattr(settings, "MULTIPLE_APP_SERVERS", None):
-        for server in settings.MULTIPLE_APP_SERVERS:
-            ret = os.system("ssh %s@%s %s" % (SYNC_USER, server, command))
-            if ret != 0:
-                ret_val = ret
-        return ret_val
-    ret = os.system(command)
-    return ret
-
-
-def broadcast(type, task, args):  # pylint: disable=redefined-builtin
+def broadcast(type, task, args, kwargs=None):  # pylint: disable=redefined-builtin
     assert type in ['web', 'app', 'build']
+    if kwargs is None:
+        kwargs = {}
     default_queue = getattr(settings, 'CELERY_DEFAULT_QUEUE', 'celery')
     if type in ['web', 'app']:
         servers = getattr(settings, "MULTIPLE_APP_SERVERS", [default_queue])
@@ -53,6 +41,7 @@ def broadcast(type, task, args):  # pylint: disable=redefined-builtin
         task.apply_async(
             queue=server,
             args=args,
+            kwargs=kwargs,
         )
 
 

readthedocs/projects/models.py

@@ -25,7 +25,7 @@
 from readthedocs.projects import constants
 from readthedocs.projects.exceptions import ProjectImportError
 from readthedocs.projects.templatetags.projects_tags import sort_version_aware
-from readthedocs.projects.utils import make_api_version, update_static_metadata
+from readthedocs.projects.utils import make_api_version
 from readthedocs.projects.version_handling import determine_stable_version
 from readthedocs.projects.version_handling import version_windows
 from readthedocs.core.resolver import resolve, resolve_domain
@@ -330,7 +330,8 @@ def save(self, *args, **kwargs):  # pylint: disable=arguments-differ
         except Exception:
             log.error('failed to symlink project', exc_info=True)
         try:
-            update_static_metadata(project_pk=self.pk)
+            if not first_save:
+                broadcast(type='app', task=tasks.update_static_metadata, args=[self.pk])
         except Exception:
             log.error('failed to update static metadata', exc_info=True)
         try:

readthedocs/projects/tasks.py

@@ -32,7 +32,7 @@
                                           BUILD_STATE_FINISHED)
 from readthedocs.builds.models import Build, Version
 from readthedocs.builds.signals import build_complete
-from readthedocs.core.utils import send_email, run_on_app_servers, broadcast
+from readthedocs.core.utils import send_email, broadcast
 from readthedocs.core.symlink import PublicSymlink, PrivateSymlink
 from readthedocs.cdn.purge import purge
 from readthedocs.doc_builder.loader import get_builder_class
@@ -395,11 +395,10 @@ def build_docs_html(self):
 
         # Gracefully attempt to move files via task on web workers.
         try:
-            move_files.delay(
-                version_pk=self.version.pk,
-                html=True,
-                hostname=socket.gethostname(),
-            )
+            broadcast(type='app', task=move_files,
+                      args=[self.version.pk, socket.gethostname()],
+                      kwargs=dict(html=True)
+                      )
         except socket.error:
             # TODO do something here
             pass
@@ -555,25 +554,27 @@ def finish_build(version_pk, build_pk, hostname=None, html=False,
         version.save()
 
     if not pdf:
-        clear_pdf_artifacts(version)
+        broadcast(type='app', task=clear_pdf_artifacts, args=[version.pk])
     if not epub:
-        clear_epub_artifacts(version)
-
-    move_files(
-        version_pk=version_pk,
-        hostname=hostname,
-        html=html,
-        localmedia=localmedia,
-        search=search,
-        pdf=pdf,
-        epub=epub,
-    )
+        broadcast(type='app', task=clear_epub_artifacts, args=[version.pk])
+
+    # Sync files to the web servers
+    broadcast(type='app', task=move_files, args=[version_pk, hostname],
+              kwargs=dict(
+                  html=html,
+                  localmedia=localmedia,
+                  search=search,
+                  pdf=pdf,
+                  epub=epub,
+    ))
 
     # Symlink project on every web
     broadcast(type='app', task=symlink_project, args=[version.project.pk])
 
+    # Update metadata
+    broadcast(type='app', task=update_static_metadata, args=[version.project.pk])
+
     # Delayed tasks
-    update_static_metadata.delay(version.project.pk)
     fileify.delay(version.pk, commit=build.commit)
     update_search.delay(version.pk, commit=build.commit)
 
@@ -887,7 +888,6 @@ def update_static_metadata(project_pk, path=None):
         fh = open(path, 'w+')
         json.dump(metadata, fh)
         fh.close()
-        Syncer.copy(path, path, host=socket.gethostname(), is_file=True)
     except (AttributeError, IOError) as e:
         log.debug(LOG_TEMPLATE.format(
             project=project.slug,
@@ -909,7 +909,7 @@ def remove_dir(path):
     shutil.rmtree(path, ignore_errors=True)
 
 
-@task(queue='web')
+@task()
 def clear_artifacts(version_pk):
     """Remove artifacts from the web servers"""
     version = Version.objects.get(pk=version_pk)
@@ -920,33 +920,19 @@ def clear_artifacts(version_pk):
 
 
 def clear_pdf_artifacts(version):
-    run_on_app_servers('rm -rf %s'
-                       % version.project.get_production_media_path(
-                           type_='pdf', version_slug=version.slug))
+    remove_dir(version.project.get_production_media_path(
+        type_='pdf', version_slug=version.slug))
 
 
 def clear_epub_artifacts(version):
-    run_on_app_servers('rm -rf %s'
-                       % version.project.get_production_media_path(
-                           type_='epub', version_slug=version.slug))
+    remove_dir(version.project.get_production_media_path(
+        type_='epub', version_slug=version.slug))
 
 
 def clear_htmlzip_artifacts(version):
-    run_on_app_servers('rm -rf %s'
-                       % version.project.get_production_media_path(
-                           type_='htmlzip', version_slug=version.slug))
+    remove_dir(version.project.get_production_media_path(
+        type_='htmlzip', version_slug=version.slug))
 
 
 def clear_html_artifacts(version):
-    run_on_app_servers('rm -rf %s' % version.project.rtd_build_path(version=version.slug))
-
-
-@task(queue='web')
-def remove_path_from_web(path):
-    """Remove the given path from the web servers file system."""
-    # Santity check  for spaces in the path since spaces would result in
-    # deleting unpredictable paths with "rm -rf".
-    assert ' ' not in path, "No spaces allowed in path"
-
-    # TODO: We need some proper escaping here for the given path.
-    run_on_app_servers('rm -rf {path}'.format(path=path))
+    remove_dir(version.project.rtd_build_path(version=version.slug))

readthedocs/projects/utils.py

@@ -29,12 +29,6 @@ def version_from_slug(slug, version):
     return v
 
 
-def update_static_metadata(project_pk):
-    """This is here to avoid circular imports in models.py"""
-    from readthedocs.projects import tasks
-    tasks.update_static_metadata.delay(project_pk)
-
-
 def find_file(filename):
     """Recursively find matching file from the current working path
 

readthedocs/rtd_tests/tests/test_celery.py

@@ -1,4 +1,3 @@
-from __future__ import absolute_import
 import os
 import json
 import shutil

readthedocs/rtd_tests/tests/test_core_tags.py

@@ -14,7 +14,7 @@ class CoreTagsTests(TestCase):
     fixtures = ["eric", "test_data"]
 
     def setUp(self):
-        with mock.patch('readthedocs.projects.models.update_static_metadata'):
+        with mock.patch('readthedocs.projects.models.broadcast'):
             self.client.login(username='eric', password='test')
             self.pip = Project.objects.get(slug='pip')
             self.pip_fr = Project.objects.create(name="PIP-FR", slug='pip-fr', language='fr', main_language_project=self.pip)

readthedocs/rtd_tests/tests/test_project_views.py

@@ -21,6 +21,7 @@
 from readthedocs.projects.models import Project, Domain
 from readthedocs.projects.views.private import ImportWizardView
 from readthedocs.projects.views.mixins import ProjectRelationMixin
+from readthedocs.projects import tasks
 
 
 @patch('readthedocs.projects.views.private.trigger_build', lambda x, basic: None)
@@ -371,13 +372,13 @@ def test_delete_project(self):
         response = self.client.get('/dashboard/pip/delete/')
         self.assertEqual(response.status_code, 200)
 
-        patcher = patch('readthedocs.projects.tasks.remove_dir')
-        with patcher as remove_dir:
+        with patch('readthedocs.projects.views.private.broadcast') as broadcast:
             response = self.client.post('/dashboard/pip/delete/')
             self.assertEqual(response.status_code, 302)
             self.assertFalse(Project.objects.filter(slug='pip').exists())
-            remove_dir.apply_async.assert_called_with(
-                queue='celery',
+            broadcast.assert_called_with(
+                type='app',
+                task=tasks.remove_dir,
                 args=[project.doc_path])
 
 

readthedocs/rtd_tests/tests/test_resolver.py

@@ -17,14 +17,13 @@ class ResolverBase(TestCase):
 
     def setUp(self):
         with mock.patch('readthedocs.projects.models.broadcast'):
-            with mock.patch('readthedocs.projects.models.update_static_metadata'):
-                self.owner = create_user(username='owner', password='test')
-                self.tester = create_user(username='tester', password='test')
-                self.pip = get(Project, slug='pip', users=[self.owner], main_language_project=None)
-                self.subproject = get(Project, slug='sub', language='ja', users=[self.owner], main_language_project=None)
-                self.translation = get(Project, slug='trans', language='ja', users=[self.owner], main_language_project=None)
-                self.pip.add_subproject(self.subproject)
-                self.pip.translations.add(self.translation)
+            self.owner = create_user(username='owner', password='test')
+            self.tester = create_user(username='tester', password='test')
+            self.pip = get(Project, slug='pip', users=[self.owner], main_language_project=None)
+            self.subproject = get(Project, slug='sub', language='ja', users=[self.owner], main_language_project=None)
+            self.translation = get(Project, slug='trans', language='ja', users=[self.owner], main_language_project=None)
+            self.pip.add_subproject(self.subproject)
+            self.pip.translations.add(self.translation)
 
 
 class SmartResolverPathTests(ResolverBase):

readthedocs/rtd_tests/tests/test_subprojects.py

@@ -68,16 +68,15 @@ class ResolverBase(TestCase):
 
     def setUp(self):
         with mock.patch('readthedocs.projects.models.broadcast'):
-            with mock.patch('readthedocs.projects.models.update_static_metadata'):
-                self.owner = create_user(username='owner', password='test')
-                self.tester = create_user(username='tester', password='test')
-                self.pip = get(Project, slug='pip', users=[self.owner], main_language_project=None)
-                self.subproject = get(Project, slug='sub', language='ja', users=[
-                                      self.owner], main_language_project=None)
-                self.translation = get(Project, slug='trans', language='ja', users=[
-                                       self.owner], main_language_project=None)
-                self.pip.add_subproject(self.subproject)
-                self.pip.translations.add(self.translation)
+            self.owner = create_user(username='owner', password='test')
+            self.tester = create_user(username='tester', password='test')
+            self.pip = get(Project, slug='pip', users=[self.owner], main_language_project=None)
+            self.subproject = get(Project, slug='sub', language='ja', users=[
+                                  self.owner], main_language_project=None)
+            self.translation = get(Project, slug='trans', language='ja', users=[
+                                   self.owner], main_language_project=None)
+            self.pip.add_subproject(self.subproject)
+            self.pip.translations.add(self.translation)
 
     @override_settings(PRODUCTION_DOMAIN='readthedocs.org')
     def test_resolver_subproject_alias(self):