move commit fetching to celery task (#5275)

* move commit fetching to celery task

* fix tests

Author: macqueen

src/sentry/api/endpoints/organization_release_details.py

@@ -167,7 +167,7 @@ def put(self, request, organization, version):
                     'refs': ['You must use an authenticated API token to fetch refs']
                 }, status=400)
             fetch_commits = not commit_list
-            release.set_refs(refs, request.user, fetch_commits=fetch_commits)
+            release.set_refs(refs, request.user, fetch=fetch_commits)
 
         if (not was_released and release.date_released):
             for project in release.projects.all():

src/sentry/api/endpoints/organization_releases.py

@@ -189,7 +189,7 @@ def post(self, request, organization):
                         'refs': ['You must use an authenticated API token to fetch refs']
                     }, status=400)
                 fetch_commits = not commit_list
-                release.set_refs(refs, request.user, fetch_commits=fetch_commits)
+                release.set_refs(refs, request.user, fetch=fetch_commits)
 
             if not created and not new_projects:
                 # This is the closest status code that makes sense, and we want

src/sentry/conf/server.py

@@ -445,6 +445,7 @@ def env(key, default='', type=None):
     'sentry.tasks.auth',
     'sentry.tasks.auto_resolve_issues',
     'sentry.tasks.beacon',
+    'sentry.tasks.commits',
     'sentry.tasks.check_auth',
     'sentry.tasks.clear_expired_snoozes',
     'sentry.tasks.collect_project_platforms',
@@ -464,6 +465,7 @@ def env(key, default='', type=None):
 CELERY_QUEUES = [
     Queue('alerts', routing_key='alerts'),
     Queue('auth', routing_key='auth'),
+    Queue('commits', routing_key='commits'),
     Queue('cleanup', routing_key='cleanup'),
     Queue('default', routing_key='default'),
     Queue('digests.delivery', routing_key='digests.delivery'),

src/sentry/models/release.py

@@ -9,7 +9,6 @@
 
 import logging
 import re
-import six
 
 from django.db import models, IntegrityError, transaction
 from django.db.models import F
@@ -19,7 +18,6 @@
 from sentry.db.models import (
     BoundedPositiveIntegerField, FlexibleForeignKey, Model, sane_repr
 )
-from sentry.exceptions import InvalidIdentity, PluginError
 from sentry.utils.cache import cache
 from sentry.utils.hashlib import md5_text
 
@@ -247,9 +245,9 @@ def add_project(self, project):
         else:
             return True
 
-    def set_refs(self, refs, user, fetch_commits=False):
+    def set_refs(self, refs, user, fetch=False):
         from sentry.models import Commit, ReleaseHeadCommit, Repository
-        from sentry.plugins import bindings
+        from sentry.tasks.commits import fetch_commits
 
         # TODO: this does the wrong thing unless you are on the most
         # recent release.  Add a timestamp compare?
@@ -258,8 +256,6 @@ def set_refs(self, refs, user, fetch_commits=False):
             projects__in=self.projects.all(),
         ).exclude(version=self.version).order_by('-date_added').first()
 
-        commit_list = []
-
         for ref in refs:
             try:
                 repo = Repository.objects.get(
@@ -283,43 +279,15 @@ def set_refs(self, refs, user, fetch_commits=False):
                     'commit': commit,
                 }
             )
-            if fetch_commits:
-                try:
-                    provider_cls = bindings.get('repository.provider').get(repo.provider)
-                except KeyError:
-                    continue
-
-                # if previous commit isn't provided, try to get from
-                # previous release otherwise, give up
-                if ref.get('previousCommit'):
-                    start_sha = ref['previousCommit']
-                elif prev_release:
-                    try:
-                        start_sha = Commit.objects.filter(
-                            organization_id=self.organization_id,
-                            releaseheadcommit__release=prev_release,
-                            repository_id=repo.id,
-                        ).values_list('key', flat=True)[0]
-                    except IndexError:
-                        continue
-                else:
-                    continue
-
-                end_sha = commit.key
-                provider = provider_cls(id=repo.provider)
-                try:
-                    repo_commits = provider.compare_commits(
-                        repo, start_sha, end_sha, actor=user
-                    )
-                except NotImplementedError:
-                    pass
-                except (PluginError, InvalidIdentity) as e:
-                    logger.exception(six.text_type(e))
-                else:
-                    commit_list.extend(repo_commits)
-
-            if commit_list:
-                self.set_commits(commit_list)
+        if fetch:
+            fetch_commits.apply_async(
+                kwargs={
+                    'release_id': self.id,
+                    'user_id': user.id,
+                    'refs': refs,
+                    'prev_release_id': prev_release and prev_release.id,
+                }
+            )
 
     def set_commits(self, commit_list):
         from sentry.models import (

src/sentry/tasks/commits.py

@@ -0,0 +1,83 @@
+from __future__ import absolute_import
+
+import logging
+import six
+
+from sentry.exceptions import InvalidIdentity, PluginError
+from sentry.models import Commit, Release, Repository, User
+from sentry.plugins import bindings
+from sentry.tasks.base import instrumented_task, retry
+
+logger = logging.getLogger(__name__)
+
+
+@instrumented_task(name='sentry.tasks.commit.fetch_commits', queue='commits',
+                   default_retry_delay=60 * 5, max_retries=5)
+@retry(exclude=(Release.DoesNotExist, User.DoesNotExist,))
+def fetch_commits(release_id, user_id, refs, prev_release_id=None, **kwargs):
+    commit_list = []
+
+    release = Release.objects.get(id=release_id)
+    user = User.objects.get(id=user_id)
+
+    prev_release = None
+    if prev_release_id is not None:
+        try:
+            prev_release = Release.objects.get(id=prev_release_id)
+        except Release.DoesNotExist:
+            pass
+
+    for ref in refs:
+        try:
+            repo = Repository.objects.get(
+                organization_id=release.organization_id,
+                name=ref['repository'],
+            )
+        except Repository.DoesNotExist:
+            continue
+
+        try:
+            commit = Commit.objects.get(
+                organization_id=release.organization_id,
+                repository_id=repo.id,
+                key=ref['commit'],
+            )
+        except Commit.DoesNotExist:
+            continue
+
+        try:
+            provider_cls = bindings.get('repository.provider').get(repo.provider)
+        except KeyError:
+            continue
+
+        # if previous commit isn't provided, try to get from
+        # previous release otherwise, give up
+        if ref.get('previousCommit'):
+            start_sha = ref['previousCommit']
+        elif prev_release:
+            try:
+                start_sha = Commit.objects.filter(
+                    organization_id=release.organization_id,
+                    releaseheadcommit__release=prev_release,
+                    repository_id=repo.id,
+                ).values_list('key', flat=True)[0]
+            except IndexError:
+                continue
+        else:
+            continue
+
+        end_sha = commit.key
+        provider = provider_cls(id=repo.provider)
+        try:
+            repo_commits = provider.compare_commits(
+                repo, start_sha, end_sha, actor=user
+            )
+        except NotImplementedError:
+            pass
+        except (PluginError, InvalidIdentity) as e:
+            logger.exception(six.text_type(e))
+        else:
+            commit_list.extend(repo_commits)
+
+    if commit_list:
+        release.set_commits(commit_list)

tests/sentry/api/endpoints/test_organization_release_details.py

@@ -1,5 +1,6 @@
 from __future__ import absolute_import
 
+from mock import patch
 from datetime import datetime
 from django.core.urlresolvers import reverse
 
@@ -94,7 +95,8 @@ def test_multiple_projects(self):
 
 
 class UpdateReleaseDetailsTest(APITestCase):
-    def test_simple(self):
+    @patch('sentry.tasks.commits.fetch_commits')
+    def test_simple(self, mock_fetch_commits):
         user = self.create_user(is_staff=False, is_superuser=False)
         org = self.organization
         org.flags.allow_joinleave = False
@@ -161,28 +163,20 @@ def test_simple(self):
             ],
         })
 
+        mock_fetch_commits.apply_async.assert_called_with(
+            kwargs={
+                'release_id': release.id,
+                'user_id': user.id,
+                'refs': [
+                    {'commit': 'a' * 40, 'repository': repo.name},
+                    {'commit': 'b' * 40, 'repository': repo2.name},
+                ],
+                'prev_release_id': base_release.id,
+            }
+        )
+
         assert response.status_code == 200, response.content
         assert response.data['version'] == release.version
-        assert ReleaseCommit.objects.filter(
-            commit__repository_id=repo.id,
-            commit__key='62de626b7c7cfb8e77efb4273b1a3df4123e6216',
-            release__version=response.data['version'],
-        ).exists()
-        assert ReleaseCommit.objects.filter(
-            commit__repository_id=repo.id,
-            commit__key='58de626b7c7cfb8e77efb4273b1a3df4123e6345',
-            release__version=response.data['version'],
-        ).exists()
-        assert ReleaseCommit.objects.filter(
-            commit__repository_id=repo2.id,
-            commit__key='62de626b7c7cfb8e77efb4273b1a3df4123e6216',
-            release__version=response.data['version'],
-        ).exists()
-        assert ReleaseCommit.objects.filter(
-            commit__repository_id=repo2.id,
-            commit__key='58de626b7c7cfb8e77efb4273b1a3df4123e6345',
-            release__version=response.data['version'],
-        ).exists()
 
         release = Release.objects.get(id=release.id)
         assert release.ref == 'master'
@@ -195,7 +189,8 @@ def test_simple(self):
         response = self.client.put(url, {'ref': 'master'})
         assert response.status_code == 403
 
-    def test_deprecated_head_commits(self):
+    @patch('sentry.tasks.commits.fetch_commits')
+    def test_deprecated_head_commits(self, mock_fetch_commits):
         user = self.create_user(is_staff=False, is_superuser=False)
         org = self.organization
         org.flags.allow_joinleave = False
@@ -263,28 +258,20 @@ def test_deprecated_head_commits(self):
             ],
         })
 
+        mock_fetch_commits.apply_async.assert_called_with(
+            kwargs={
+                'release_id': release.id,
+                'user_id': user.id,
+                'refs': [
+                    {'commit': 'a' * 40, 'previousCommit': None, 'repository': repo.name},
+                    {'commit': 'b' * 40, 'previousCommit': None, 'repository': repo2.name},
+                ],
+                'prev_release_id': base_release.id,
+            }
+        )
+
         assert response.status_code == 200, response.content
         assert response.data['version'] == release.version
-        assert ReleaseCommit.objects.filter(
-            commit__repository_id=repo.id,
-            commit__key='62de626b7c7cfb8e77efb4273b1a3df4123e6216',
-            release__version=response.data['version'],
-        ).exists()
-        assert ReleaseCommit.objects.filter(
-            commit__repository_id=repo.id,
-            commit__key='58de626b7c7cfb8e77efb4273b1a3df4123e6345',
-            release__version=response.data['version'],
-        ).exists()
-        assert ReleaseCommit.objects.filter(
-            commit__repository_id=repo2.id,
-            commit__key='62de626b7c7cfb8e77efb4273b1a3df4123e6216',
-            release__version=response.data['version'],
-        ).exists()
-        assert ReleaseCommit.objects.filter(
-            commit__repository_id=repo2.id,
-            commit__key='58de626b7c7cfb8e77efb4273b1a3df4123e6345',
-            release__version=response.data['version'],
-        ).exists()
 
         release = Release.objects.get(id=release.id)
         assert release.ref == 'master'