Add events workflow (#130)

Author: La0

.isort.cfg

@@ -1,6 +1,6 @@
 [settings]
-known_first_party = code_coverage_backend,code_coverage_bot,code_coverage_tools,conftest
-known_third_party = connexion,datadog,dateutil,fakeredis,flask,flask_cors,flask_talisman,google,hglib,jsone,jsonschema,libmozdata,logbook,pytest,pytz,raven,redis,requests,responses,setuptools,structlog,taskcluster,werkzeug,zstandard
+known_first_party = code_coverage_backend,code_coverage_bot,code_coverage_events,code_coverage_tools,conftest
+known_third_party = connexion,datadog,dateutil,fakeredis,flask,flask_cors,flask_talisman,google,hglib,jsone,jsonschema,libmozdata,libmozevent,logbook,pytest,pytz,raven,redis,requests,responses,setuptools,structlog,taskcluster,werkzeug,zstandard
 force_single_line = True
 default_section=FIRSTPARTY
 line_length=159

.taskcluster.yml

@@ -116,6 +116,26 @@ tasks:
         owner: [email protected]
         source: https://github.com/mozilla/code-coverage
 
+    - taskId: {$eval: as_slugid("events_check_tests")}
+      provisionerId: aws-provisioner-v1
+      workerType: github-worker
+      created: {$fromNow: ''}
+      deadline: {$fromNow: '1 hour'}
+      payload:
+        maxRunTime: 3600
+        image: python:3
+        command:
+          - sh
+          - -lxce
+          - "git clone --quiet ${repository} /src && cd /src && git checkout ${head_rev} -b checks &&
+            cd /src/events && pip install -q -r requirements.txt && pip install --quiet . && pip install --quiet -r requirements-dev.txt &&
+            pytest -v"
+      metadata:
+        name: "Code Coverage Events checks: unit tests"
+        description: Check python code with pytest
+        owner: [email protected]
+        source: https://github.com/mozilla/code-coverage
+
     - taskId: {$eval: as_slugid("backend_build")}
       created: {$fromNow: ''}
       deadline: {$fromNow: '1 hour'}
@@ -198,6 +218,47 @@ tasks:
         owner: [email protected]
         source: https://github.com/mozilla/code-coverage
 
+    - taskId: {$eval: as_slugid("events_build")}
+      created: {$fromNow: ''}
+      deadline: {$fromNow: '1 hour'}
+      provisionerId: aws-provisioner-v1
+      workerType: releng-svc
+      dependencies:
+        - {$eval: as_slugid("check_lint")}
+        - {$eval: as_slugid("events_check_tests")}
+      payload:
+        capabilities:
+          privileged: true
+        maxRunTime: 3600
+        image: "${taskboot_image}"
+        env:
+          GIT_REPOSITORY: ${repository}
+          GIT_REVISION: ${head_rev}
+        command:
+          - taskboot
+          - build
+          - --image
+          - mozilla/code-coverage
+          - --tag
+          - "events-${channel}"
+          - --tag
+          - "events-${head_rev}"
+          - --write
+          - /events.tar
+          - events/Dockerfile
+        artifacts:
+          public/code-coverage-events.tar:
+            expires: {$fromNow: '2 weeks'}
+            path: /events.tar
+            type: file
+      scopes:
+        - docker-worker:capability:privileged
+      metadata:
+        name: Code Coverage Events docker build
+        description: Build docker image with taskboot
+        owner: [email protected]
+        source: https://github.com/mozilla/code-coverage
+
     - taskId: {$eval: as_slugid("addon_build")}
       provisionerId: aws-provisioner-v1
       workerType: github-worker
@@ -282,6 +343,36 @@ tasks:
           owner: [email protected]
           source: https://github.com/mozilla/code-coverage
 
+    - $if: 'channel in ["testing", "production"]'
+      then:
+        taskId: {$eval: as_slugid("events_deploy")}
+        created: {$fromNow: ''}
+        deadline: {$fromNow: '1 hour'}
+        provisionerId: aws-provisioner-v1
+        workerType: github-worker
+        dependencies:
+          - {$eval: as_slugid("events_build")}
+        payload:
+          features:
+            taskclusterProxy: true
+          maxRunTime: 3600
+          image: "${taskboot_image}"
+          command:
+            - taskboot
+            - deploy-heroku
+            - --heroku-app
+            - "code-coverage-events-${channel}"
+            - worker:public/code-coverage-events.tar
+          env:
+            TASKCLUSTER_SECRET: "project/relman/code-coverage/deploy-${channel}"
+        scopes:
+          - "secrets:get:project/relman/code-coverage/deploy-${channel}"
+        metadata:
+          name: "Code Coverage Events deployment (${channel})"
+          description: Deploy docker image on Heroku
+          owner: [email protected]
+          source: https://github.com/mozilla/code-coverage
+
     - $if: 'channel in ["testing", "production"]'
       then:
         taskId: {$eval: as_slugid("bot_deploy")}

events/Dockerfile

@@ -0,0 +1,7 @@
+FROM python:3-slim
+
+ADD events /src
+
+RUN cd /src && pip install --disable-pip-version-check --no-cache-dir -r requirements.txt && python setup.py install
+
+CMD ["code-coverage-events"]

events/VERSION

@@ -0,0 +1 @@
+1.0.0

events/code_coverage_events/__init__.py

@@ -0,0 +1,4 @@
+# -*- coding: utf-8 -*-
+
+QUEUE_MONITORING = "monitoring"
+QUEUE_PULSE = "pulse"

events/code_coverage_events/cli.py

@@ -0,0 +1,51 @@
+# -*- coding: utf-8 -*-
+import argparse
+import os
+
+import structlog
+from libmozevent import taskcluster_config
+from libmozevent.log import init_logger
+
+from code_coverage_events.workflow import Events
+
+logger = structlog.get_logger(__name__)
+
+
+def parse_cli():
+    """
+    Setup CLI options parser
+    """
+    parser = argparse.ArgumentParser(description="Mozilla Code Review Bot")
+    parser.add_argument(
+        "--taskcluster-secret",
+        help="Taskcluster Secret path",
+        default=os.environ.get("TASKCLUSTER_SECRET"),
+    )
+    parser.add_argument("--taskcluster-client-id", help="Taskcluster Client ID")
+    parser.add_argument("--taskcluster-access-token", help="Taskcluster Access token")
+    return parser.parse_args()
+
+
+def main():
+    args = parse_cli()
+    taskcluster_config.auth(args.taskcluster_client_id, args.taskcluster_access_token)
+    taskcluster_config.load_secrets(
+        args.taskcluster_secret,
+        "events",
+        required=("pulse_user", "pulse_password", "hook_id", "hook_group_id"),
+        existing=dict(admins=["[email protected]", "[email protected]"]),
+    )
+
+    init_logger(
+        "code_coverage_events",
+        PAPERTRAIL_HOST=taskcluster_config.secrets.get("PAPERTRAIL_HOST"),
+        PAPERTRAIL_PORT=taskcluster_config.secrets.get("PAPERTRAIL_PORT"),
+        SENTRY_DSN=taskcluster_config.secrets.get("SENTRY_DSN"),
+    )
+
+    events = Events()
+    events.run()
+
+
+if __name__ == "__main__":
+    main()

events/code_coverage_events/workflow.py

@@ -0,0 +1,187 @@
+# -*- coding: utf-8 -*-
+
+import asyncio
+
+import requests
+import structlog
+from libmozevent import taskcluster_config
+from libmozevent.bus import MessageBus
+from libmozevent.monitoring import Monitoring
+from libmozevent.pulse import PulseListener
+from libmozevent.utils import retry
+from libmozevent.utils import run_tasks
+
+from code_coverage_events import QUEUE_MONITORING
+from code_coverage_events import QUEUE_PULSE
+
+logger = structlog.get_logger(__name__)
+
+
+class CodeCoverage(object):
+    """
+    Taskcluster hook handling the code coverage
+    """
+
+    def __init__(self, hook_id, hook_group_id, bus):
+        self.triggered_groups = set()
+        self.group_id = hook_group_id
+        self.hook_id = hook_id
+        self.bus = bus
+
+        # Setup TC services
+        self.queue = taskcluster_config.get_service("queue")
+        self.hooks = taskcluster_config.get_service("hooks")
+
+    async def run(self):
+        """
+        Main consumer, running queued payloads from the pulse listener
+        """
+        while True:
+            # Get next payload from pulse messages
+            payload = await self.bus.receive(QUEUE_PULSE)
+
+            # Parse the payload to extract a new task's environment
+            envs = await self.parse(payload)
+            if envs is None:
+                continue
+
+            for env in envs:
+                # Trigger new tasks
+                task = self.hooks.triggerHook(self.group_id, self.hook_id, env)
+                task_id = task["status"]["taskId"]
+                logger.info("Triggered a new code coverage task", id=task_id)
+
+                # Send task to monitoring
+                await self.bus.send(
+                    QUEUE_MONITORING, (self.group_id, self.hook_id, task_id)
+                )
+
+    def is_coverage_task(self, task):
+        return any(
+            task["task"]["metadata"]["name"].startswith(s)
+            for s in ["build-linux64-ccov", "build-win64-ccov"]
+        )
+
+    async def get_build_task_in_group(self, group_id):
+        if group_id in self.triggered_groups:
+            logger.info(
+                "Received duplicated groupResolved notification", group=group_id
+            )
+            return None
+
+        def maybe_trigger(tasks):
+            logger.info(
+                "Checking code coverage tasks", group_id=group_id, nb=len(tasks)
+            )
+            for task in tasks:
+                if self.is_coverage_task(task):
+                    self.triggered_groups.add(group_id)
+                    return task
+
+            return None
+
+        def load_tasks(limit=200, continuationToken=None):
+            query = {"limit": limit}
+            if continuationToken is not None:
+                query["continuationToken"] = continuationToken
+            reply = retry(lambda: self.queue.listTaskGroup(group_id, query=query))
+            return maybe_trigger(reply["tasks"]), reply.get("continuationToken")
+
+        async def retrieve_coverage_task():
+            task, token = load_tasks()
+
+            while task is None and token is not None:
+                task, token = load_tasks(continuationToken=token)
+
+                # Let other tasks run on long batches
+                await asyncio.sleep(0)
+
+            return task
+
+        try:
+            return await retrieve_coverage_task()
+        except requests.exceptions.HTTPError:
+            return None
+
+    async def parse(self, body):
+        """
+        Extract revisions from payload
+        """
+        taskGroupId = body["taskGroupId"]
+
+        build_task = await self.get_build_task_in_group(taskGroupId)
+        if build_task is None:
+            return None
+
+        repository = build_task["task"]["payload"]["env"]["GECKO_HEAD_REPOSITORY"]
+
+        if repository not in [
+            "https://hg.mozilla.org/mozilla-central",
+            "https://hg.mozilla.org/try",
+        ]:
+            logger.warn(
+                "Received groupResolved notification for a coverage task in an unexpected branch",
+                repository=repository,
+            )
+            return None
+
+        logger.info(
+            "Received groupResolved notification for coverage builds",
+            repository=repository,
+            revision=build_task["task"]["payload"]["env"]["GECKO_HEAD_REV"],
+            group=taskGroupId,
+        )
+
+        return [
+            {
+                "REPOSITORY": repository,
+                "REVISION": build_task["task"]["payload"]["env"]["GECKO_HEAD_REV"],
+            }
+        ]
+
+
+class Events(object):
+    """
+    Listen to pulse events and trigger new code coverage tasks
+    """
+
+    def __init__(self):
+        # Create message bus shared amongst process
+        self.bus = MessageBus()
+
+        # Build code coverage workflow
+        self.workflow = CodeCoverage(
+            taskcluster_config.secrets["hook_id"],
+            taskcluster_config.secrets["hook_group_id"],
+            self.bus,
+        )
+
+        # Setup monitoring for newly created tasks
+        self.monitoring = Monitoring(
+            QUEUE_MONITORING, taskcluster_config.secrets["admins"], 7 * 3600
+        )
+        self.monitoring.register(self.bus)
+
+        # Create pulse listener for code coverage
+        self.pulse = PulseListener(
+            QUEUE_PULSE,
+            "exchange/taskcluster-queue/v1/task-group-resolved",
+            "#",
+            taskcluster_config.secrets["pulse_user"],
+            taskcluster_config.secrets["pulse_password"],
+        )
+        self.pulse.register(self.bus)
+
+    def run(self):
+
+        consumers = [
+            # Code coverage main workflow
+            self.workflow.run(),
+            # Add monitoring task
+            self.monitoring.run(),
+            # Add pulse task
+            self.pulse.run(),
+        ]
+
+        # Run all tasks concurrently
+        run_tasks(consumers)

events/requirements-dev.txt

@@ -0,0 +1,3 @@
+pytest
+pytest-asyncio
+responses

events/requirements.txt

@@ -0,0 +1 @@
+libmozevent==1.0.0