Bug 1622336 - Trigger schedules API as soon as a push happens (#1548)

This change adds a Pulse listener for hg pushes as a background process
to `bugbug-http-service`.

This listener works if `PULSE_USER` and `PULSE_PASSWORD` are set before starting
up the Docker containers.

The listener only pays attention to autoland and try pushes. When these happen
we do a GET request from the bugbug deployment. This will get the schedules service
ready for when the Gecko decision task will query it.

.isort.cfg

@@ -3,4 +3,4 @@ multi_line_output=3
 include_trailing_comma=True
 line_length=88
 known_first_party = bugbug,bugbug_http
-known_third_party = adr,apispec,apispec_webframeworks,boto3,cerberus,dateutil,flask,flask_cors,hglib,imblearn,joblib,jsone,jsonschema,libmozdata,lmdb,marshmallow,matplotlib,microannotate,mozci,numpy,orjson,pandas,pkg_resources,pyemd,pytest,redis,requests,responses,rq,scipy,setuptools,shap,sklearn,tabulate,taskcluster,tenacity,tqdm,xgboost,yaml,zstandard
+known_third_party = adr,apispec,apispec_webframeworks,boto3,cerberus,dateutil,flask,flask_cors,hglib,imblearn,joblib,jsone,jsonschema,kombu,libmozdata,lmdb,marshmallow,matplotlib,microannotate,mozci,numpy,orjson,pandas,pkg_resources,pyemd,pytest,redis,requests,responses,rq,scipy,setuptools,shap,sklearn,tabulate,taskcluster,tenacity,tqdm,xgboost,yaml,zstandard

http_service/Dockerfile

@@ -10,4 +10,5 @@ RUN pip install --disable-pip-version-check --quiet --no-cache-dir -r /requireme
 COPY . /code/http_service
 RUN pip install --disable-pip-version-check --quiet --no-cache-dir /code/http_service
 
-CMD gunicorn -b 0.0.0.0:$PORT bugbug_http.app --preload --timeout 30 -w 3
+# Run the Pulse listener in the background
+CMD (bugbug-http-pulse-listener &) && gunicorn -b 0.0.0.0:$PORT bugbug_http.app --preload --timeout 30 -w 3

http_service/bugbug_http/listener.py

@@ -0,0 +1,96 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+Generate your credentials in https://pulseguardian.mozilla.org
+
+Call this script like:
+  export PULSE_USER=generated_username
+  export PULSE_PASSWORD=generated_username
+  # In case you want to hit the live server
+  export BUGBUG_HTTP_SERVER=https://bugbug.herokuapp.com
+  cd http_service && docker-compose run bugbug-http-service
+"""
+import logging
+import os
+import traceback
+
+import requests
+from kombu import Connection, Exchange, Queue
+from kombu.mixins import ConsumerMixin
+
+logging.basicConfig()
+logger = logging.getLogger()
+logger.setLevel(logging.INFO)
+
+BUGBUG_HTTP_SERVER = os.environ.get("BUGBUG_HTTP_SERVER", "http://localhost:8000")
+CONNECTION_URL = "amqp://{}:{}@pulse.mozilla.org:5671/?ssl=1"
+
+
+class _GenericConsumer(ConsumerMixin):
+    def __init__(self, connection, queues, callback):
+        self.connection = connection
+        self.queues = queues
+        self.callback = callback
+
+    def get_consumers(self, Consumer, channel):
+        return [Consumer(queues=self.queues, callbacks=[self.callback])]
+
+
+class HgPushesConsumer:
+    def __init__(self, user, password, callback):
+        self.connection = Connection(CONNECTION_URL.format(user, password))
+        self.queues = [
+            Queue(
+                name="queue/{}/pushes".format(user),
+                exchange=Exchange(
+                    "exchange/hgpushes/v2", type="topic", no_declare=True,
+                ),
+                routing_key="#",
+                durable=True,
+                auto_delete=True,
+            )
+        ]
+        self.consumer = _GenericConsumer(self.connection, self.queues, callback)
+
+    def __enter__(self):
+        return self.consumer
+
+    def __exit__(self, type, value, traceback):
+        self.connection.close()
+
+
+def _on_message(body, message):
+    try:
+        branch = body["payload"]["data"]["repo_url"].split("/")[-1]
+        rev = body["payload"]["data"]["heads"][0]
+        if branch in ["autoland", "try"]:
+            url = "{}/push/{}/{}/schedules".format(BUGBUG_HTTP_SERVER, branch, rev)
+            response = requests.get(url, headers={"X-Api-Key": "pulse_listener"})
+            if response.status_code == 202:
+                logger.info("Successfully requested {}/{}".format(branch, rev))
+            else:
+                logger.warning(
+                    "We got status: {} for: {}".format(response.status_code, url)
+                )
+    except Exception as e:
+        traceback.print_tb(e)
+    finally:
+        message.ack()
+
+
+def main():
+    # Generate user/password in https://pulseguardian.mozilla.org/
+    # Set PULSE_USER and PULSE_PASSWORD as env variables
+    user = os.environ.get("PULSE_USER")
+    password = os.environ.get("PULSE_PASSWORD")
+    if user and password:
+        with HgPushesConsumer(user, password, _on_message) as consumer:
+            consumer.run()
+    else:
+        logger.warning(
+            "The Pulse listener will be skipped unless you define PULSE_USER & PULSE_PASSWORD"
+        )
+
+
+if __name__ == "__main__":
+    main()

http_service/docker-compose.yml

@@ -8,6 +8,8 @@ services:
       - BUGBUG_BUGZILLA_TOKEN
       - REDIS_URL=redis://redis:6379/0
       - PORT=8000
+      - PULSE_USER
+      - PULSE_PASSWORD
     ports:
       - target: 8000
         published: 8000

http_service/requirements.txt

@@ -5,6 +5,8 @@ Flask==1.1.2
 flask-apispec==0.8.8
 flask-cors==3.0.8
 gunicorn==20.0.4
+kombu==4.6.8
 marshmallow==3.5.2
+requests==2.23.0
 rq==1.3.0
 rq-dashboard==0.6.1

http_service/setup.py

@@ -31,7 +31,12 @@ setup(
     packages=find_packages(),
     include_package_data=True,
     license="MPL2",
-    entry_points={"console_scripts": ["bugbug-http-worker = bugbug_http.worker:main"]},
+    entry_points={
+        "console_scripts": [
+            "bugbug-http-worker = bugbug_http.worker:main",
+            "bugbug-http-pulse-listener = bugbug_http.listener:main",
+        ]
+    },
     classifiers=[
         "Programming Language :: Python :: 3.7",
         "Programming Language :: Python :: 3 :: Only",