Bug 1387418: Handle empty push resultset from hg

This patch parses the hg push resultset and handles cases
where pushes are not present by raising an appropriate error
and logging the error correctly.

Author: moijes12

tests/etl/test_push_loader.py

@@ -11,6 +11,7 @@
     HgPushTransformer,
     PulsePushError,
     PushLoader,
+    ResultsetFetchError,
 )
 from treeherder.model.models import Push
 
@@ -218,3 +219,18 @@ def test_ingest_github_push_comma_separated_branches(
         "https://firefox-ci-tc.services.mozilla.com",
     )
     assert Push.objects.count() == expected_pushes
+
+
+def test_fetch_push_raises_on_empty_pushes(monkeypatch):
+    """Test that a ResultsetFetchError is raised when fetch_json returns a dict without 'pushes'"""
+    monkeypatch.setattr("treeherder.etl.push_loader.fetch_json", lambda url: {})
+    transformer = HgPushTransformer(
+        {
+            "payload": {
+                "repo_url": "https://hg.mozilla.org/try",
+                "pushlog_pushes": [{"push_full_json_url": "http://example"}],
+            }
+        }
+    )
+    with pytest.raises(ResultsetFetchError):
+        transformer.fetch_push("http://example", repository="try")

treeherder/etl/push_loader.py

@@ -261,7 +261,45 @@ def fetch_push(self, url, repository, sha=None):
 
         logger.debug("fetching for %s %s", repository, url)
         # there will only ever be one, with this url
-        push = list(fetch_json(url)["pushes"].values())[0]
+        try:
+            data = fetch_json(url)
+        except Exception as e:
+            logger.exception("Failed fetching push JSON from %s for %s: %s", url, repository, e)
+            try:
+                newrelic.agent.record_custom_event(
+                    "resultset_fetch_failure",
+                    {"url": url, "repository": repository, "error": str(e)},
+                )
+            except Exception:
+                # NewRelic failures should not block ingestion
+                logger.debug("NewRelic event failed for fetch error")
+            raise ResultsetFetchError(f"Failed to fecth JSON from {url}: {e}") from e
+
+        pushes = None
+        if isinstance(data, dict):
+            pushes = data.get("pushes")
+
+        if not pushes or not isinstance(pushes, dict):
+            # Log deg info and raise error
+            data_keys = list(data.keys()) if isinstance(data, dict) else None
+            logger.warning(
+                f"Malformed or empty push JSON from {url} for {repository}: data-keys={data_keys}"
+            )
+            try:
+                newrelic.agent.record_custom_event(
+                    "resultset_fetch_malformed",
+                    {"url": url, "repository": repository, "data_keys": data_keys},
+                )
+            except Exception:
+                logger.debug("NewRelic event failed for malformed data")
+            raise ResultsetFetchError(f"Malformed or empty 'pushes' for {url}")
+
+        # Safely get the first push
+        try:
+            push = next(iter(pushes.values()))
+        except Exception as e:
+            logger.exception("Failed to extract first push from pushes for {respository} {url}:{e}")
+            raise ResultsetFetchError(f"Unable to extract push from 'pushes' for {url}: {e}") from e
 
         commits = []
         # we only want to ingest the last 200 commits for each push,
@@ -285,3 +323,9 @@ def fetch_push(self, url, repository, sha=None):
 
 class PulsePushError(ValueError):
     pass
+
+
+class ResultsetFetchError(Exception):
+    """Raised when fetching or parsing a push resultset fails."""
+
+    pass