Bug 1281056 - Stop using http calls when storing job data (#1640)

tests/conftest.py

@@ -227,9 +227,7 @@ def resultset_with_three_jobs(jm, sample_data, sample_resultset, test_repository
 
 
 @pytest.fixture
-def eleven_jobs_stored(jm, sample_data, sample_resultset, test_repository, mock_log_parser):
-    """stores a list of 11 job samples"""
-
+def eleven_job_blobs(jm, sample_data, sample_resultset, test_repository, mock_log_parser):
     jm.store_result_set_data(sample_resultset)
 
     num_jobs = 11
@@ -253,8 +251,13 @@ def eleven_jobs_stored(jm, sample_data, sample_resultset, test_repository, mock_
         blobs.append(blob)
 
         resultset_index += 1
+    return blobs
 
-    jm.store_job_data(blobs)
+
+@pytest.fixture
+def eleven_jobs_stored(jm, eleven_job_blobs):
+    """stores a list of 11 job samples"""
+    jm.store_job_data(eleven_job_blobs)
 
 
 @pytest.fixture

tests/webapp/api/test_jobs_api.py

@@ -4,16 +4,17 @@
 
 from treeherder.model.models import (ExclusionProfile,
                                      JobExclusion)
+from treeherder.webapp.api.jobs import JobsViewSet
 
 
-def test_job_list(webapp, eleven_jobs_stored, jm):
+def test_job_list(webapp, eleven_jobs_stored, test_repository):
     """
     test retrieving a list of ten json blobs from the jobs-list
     endpoint.
     """
     resp = webapp.get(
         reverse("jobs-list",
-                kwargs={"project": jm.project})
+                kwargs={"project": test_repository.name})
     )
     assert resp.status_int == 200
     response_dict = resp.json
@@ -374,3 +375,16 @@ def test_list_similar_jobs(webapp, eleven_jobs_stored, jm):
     assert isinstance(similar_jobs['results'], list)
 
     assert len(similar_jobs['results']) == 3
+
+
+def test_job_create(webapp, test_repository, test_user, eleven_job_blobs, monkeypatch, jm):
+    monkeypatch.setattr(JobsViewSet, 'permission_classes', ())
+
+    url = reverse("jobs-list",
+                  kwargs={"project": jm.project})
+    client = APIClient()
+    resp = client.post(url, data=eleven_job_blobs, format="json")
+
+    assert resp.status_code == 200
+    test_job_list(webapp, None, test_repository)
+    test_job_detail(webapp, None, None, jm)

treeherder/etl/buildapi.py

@@ -1,14 +1,16 @@
 import logging
+import traceback
 from collections import defaultdict
 
+import newrelic.agent
 import simplejson as json
 from django.conf import settings
 from django.core.cache import cache
 
 from treeherder.client import TreeherderJobCollection
 from treeherder.etl import (buildbot,
-                            common,
-                            th_publisher)
+                            common)
+from treeherder.model.derived.jobs import JobsModel
 from treeherder.model.models import Datasource
 
 logger = logging.getLogger(__name__)
@@ -409,8 +411,8 @@ def run(self, revision_filter=None, project_filter=None, job_group_filter=None):
                                                        project_filter=project_filter,
                                                        job_group_filter=job_group_filter)
         if job_collections:
-            th_publisher.post_treeherder_collections(job_collections,
-                                                     chunk_size=settings.BUILDAPI_BUILDS4H_CHUNK_SIZE)
+            store_jobs(job_collections,
+                       chunk_size=settings.BUILDAPI_BUILDS4H_CHUNK_SIZE)
         cache.set(CACHE_KEYS['complete'], job_ids_seen)
         return bool(job_collections)
 
@@ -426,8 +428,8 @@ def run(self, revision_filter=None, project_filter=None, job_group_filter=None):
                                                        project_filter=project_filter,
                                                        job_group_filter=job_group_filter)
         if job_collections:
-            th_publisher.post_treeherder_collections(job_collections,
-                                                     chunk_size=settings.BUILDAPI_PENDING_CHUNK_SIZE)
+            store_jobs(job_collections,
+                       chunk_size=settings.BUILDAPI_PENDING_CHUNK_SIZE)
         cache.set(CACHE_KEYS['pending'], job_ids_seen)
         return bool(job_collections)
 
@@ -443,7 +445,27 @@ def run(self, revision_filter=None, project_filter=None, job_group_filter=None):
                                                        project_filter=project_filter,
                                                        job_group_filter=job_group_filter)
         if job_collections:
-            th_publisher.post_treeherder_collections(job_collections,
-                                                     chunk_size=settings.BUILDAPI_RUNNING_CHUNK_SIZE)
+            store_jobs(job_collections,
+                       chunk_size=settings.BUILDAPI_RUNNING_CHUNK_SIZE)
         cache.set(CACHE_KEYS['running'], job_ids_seen)
         return bool(job_collections)
+
+
+def store_jobs(job_collections, chunk_size):
+    errors = []
+    for repository, jobs in job_collections.iteritems():
+        with JobsModel(repository) as jm:
+            for collection in jobs.get_chunks(chunk_size=chunk_size):
+                try:
+                    collection.validate()
+                    jm.store_job_data(collection.get_collection_data())
+                except Exception:
+                    newrelic.agent.record_exception()
+                    errors.append({
+                        "project": repository,
+                        "collection": "job",
+                        "message": traceback.format_exc()
+                    })
+
+    if errors:
+        raise common.CollectionNotStoredException(errors)

treeherder/etl/common.py

@@ -9,6 +9,23 @@
 REVISION_SHA_RE = re.compile(r'^[a-f\d]{12,40}$', re.IGNORECASE)
 
 
+class CollectionNotStoredException(Exception):
+
+    def __init__(self, error_list, *args, **kwargs):
+        """
+        error_list contains dictionaries, each containing
+        project, url and message
+        """
+        super(CollectionNotStoredException, self).__init__(args, kwargs)
+        self.error_list = error_list
+
+    def __str__(self):
+        return "\n".join(
+            ["[{project}] Error storing {collection} data: {message}".format(
+                **error) for error in self.error_list]
+        )
+
+
 class JobDataError(ValueError):
     pass
 

treeherder/etl/pushlog.py

@@ -6,7 +6,8 @@
 from django.core.cache import cache
 
 from treeherder.client import TreeherderResultSetCollection
-from treeherder.etl.common import (fetch_json,
+from treeherder.etl.common import (CollectionNotStoredException,
+                                   fetch_json,
                                    generate_revision_hash)
 from treeherder.model.derived.jobs import JobsModel
 
@@ -152,20 +153,3 @@ def run(self, source_url, repository, changeset=None):
             cache.set("{0}:last_push_id".format(repository), last_push_id)
 
         return top_revision
-
-
-class CollectionNotStoredException(Exception):
-
-    def __init__(self, error_list, *args, **kwargs):
-        """
-        error_list contains dictionaries, each containing
-        project, url and message
-        """
-        super(CollectionNotStoredException, self).__init__(args, kwargs)
-        self.error_list = error_list
-
-    def __str__(self):
-        return "\n".join(
-            ["[{project}] Error storing {collection} data: {message}".format(
-                **error) for error in self.error_list]
-        )