sync/handlers.py

import newrelic.agent

from . import downstream
from . import log
from . import landing
from . import tc
from . import trypush
from . import update
from . import upstream
from . import worktree
from .env import Environment
from .errors import RetryableError
from .gitutils import pr_for_commit, update_repositories, gecko_repo
from .load import get_pr_sync
from .lock import SyncLock
from .notify import bugupdate
from .repos import cinnabar

from git.repo.base import Repo
from typing import Any, Dict

env = Environment()

logger = log.get_logger(__name__)


class Handler:
    def __init__(self, config):
        self.config = config

    def __call__(self, git_gecko: Repo, git_wpt: Repo, body: Dict[str, Any]) -> None:
        raise NotImplementedError


def handle_pr(git_gecko: Repo, git_wpt: Repo, event: Dict[str, Any]) -> None:
    newrelic.agent.set_transaction_name("handle_pr")
    pr_id = event["number"]
    newrelic.agent.add_custom_parameter("pr", pr_id)
    newrelic.agent.add_custom_parameter("action", event["action"])
    env.gh_wpt.load_pull(event["pull_request"])

    sync = get_pr_sync(git_gecko, git_wpt, pr_id)
    repo_update = event.get("_wptsync", {}).get("repo_update", True)

    if not sync:
        # If we don't know about this sync then it's a new thing that we should
        # set up state for
        # TODO: maybe want to create a new sync here irrespective of the event
        # type because we missed some events.
        if event["action"] == "opened":
            downstream.new_wpt_pr(git_gecko, git_wpt, event["pull_request"],
                                  repo_update=repo_update)
    else:
        if isinstance(sync, downstream.DownstreamSync):
            update_func = downstream.update_pr
        elif isinstance(sync, upstream.UpstreamSync):
            update_func = upstream.update_pr
        else:
            return

        merge_sha = (event["pull_request"]["merge_commit_sha"]
                     if event["pull_request"]["merged"] else None)
        merged_by = (event["pull_request"]["merged_by"]["login"] if merge_sha else None)
        with SyncLock.for_process(sync.process_name) as lock:
            assert isinstance(lock, SyncLock)
            with sync.as_mut(lock):
                update_func(git_gecko,
                            git_wpt,
                            sync,
                            event["action"],
                            merge_sha,
                            event["pull_request"]["base"]["sha"],
                            merged_by)


def handle_check_run(git_gecko: Repo, git_wpt: Repo, event: Dict[str, Any]) -> None:
    newrelic.agent.set_transaction_name("handle_check_run")
    if event["action"] != "completed":
        return

    check_run = event["check_run"]

    newrelic.agent.add_custom_parameter("sha", check_run["head_sha"])
    newrelic.agent.add_custom_parameter("name", check_run["name"])
    newrelic.agent.add_custom_parameter("status", check_run["status"])
    newrelic.agent.add_custom_parameter("conclusion", check_run["conclusion"])

    if check_run["name"] not in env.gh_wpt.required_checks("master"):
        logger.info("Check %s is not required" % check_run["name"])
        return

    # First check if the PR is head of any pull request
    pr_id = pr_for_commit(git_wpt, check_run["head_sha"])

    if pr_id is None:
        logger.info("Commit %s is not part of a PR" % pr_id)
        return

    repo_update = event.get("_wptsync", {}).get("repo_update", True)
    if repo_update:
        update_repositories(None, git_wpt)

    sync = get_pr_sync(git_gecko, git_wpt, pr_id)
    if not isinstance(sync, upstream.UpstreamSync):
        return

    with SyncLock.for_process(sync.process_name) as lock:
        assert isinstance(lock, SyncLock)
        with sync.as_mut(lock):
            upstream.commit_check_changed(git_gecko,
                                          git_wpt,
                                          sync)


def handle_push(git_gecko: Repo, git_wpt: Repo, event: Dict[str, Any]) -> None:
    newrelic.agent.set_transaction_name("handle_push")
    update_repositories(None, git_wpt)
    landing.wpt_push(git_gecko, git_wpt, [item["id"] for item in event["commits"]])


class GitHubHandler(Handler):
    dispatch_event = {
        "pull_request": handle_pr,
        "check_run": handle_check_run,
        "push": handle_push,
    }

    def __call__(self, git_gecko: Repo, git_wpt: Repo, body: Dict[str, Any]) -> None:
        newrelic.agent.set_transaction_name("GitHubHandler")
        handler = self.dispatch_event[body["event"]]
        newrelic.agent.add_custom_parameter("event", body["event"])
        if handler is not None:
            return handler(git_gecko, git_wpt, body["payload"])
        # TODO: other events to check if we can merge a PR
        # because of some update


class PushHandler(Handler):
    def __call__(self, git_gecko: Repo, git_wpt: Repo, body: Dict[str, Any]) -> None:
        newrelic.agent.set_transaction_name("PushHandler")
        repo = body["_meta"]["routing_key"]
        if "/" in repo:
            repo_name = repo.rsplit("/", 1)[1]
        else:
            repo_name = repo

        # Commands can override the base rev and select only certain processses
        base_rev = body.get("_wptsync", {}).get("base_rev")
        processes = body.get("_wptsync", {}).get("processes")

        # Not sure if it's ever possible to get multiple heads here in a way that
        # matters for us
        rev = body["payload"]["data"]["heads"][0]
        logger.info(f"Handling commit {rev} to repo {repo}")

        newrelic.agent.add_custom_parameter("repo", repo)
        newrelic.agent.add_custom_parameter("rev", rev)

        update_repositories(git_gecko, git_wpt, wait_gecko_commit=rev)
        try:
            git_rev = cinnabar(git_gecko).hg2git(rev)
        except ValueError:
            pass
        else:
            if gecko_repo(git_gecko, git_gecko.rev_parse(git_rev)) is None:
                logger.info("Skipping commit as it isn't in a branch we track")
                return
        if processes is None or "upstream" in processes:
            upstream.gecko_push(git_gecko, git_wpt, repo_name, rev, base_rev=base_rev)
        if processes is None or "landing" in processes:
            landing.gecko_push(git_gecko, git_wpt, repo_name, rev, base_rev=base_rev)


class DecisionTaskHandler(Handler):
    """Handler for the task associated with a task completing."""

    complete_states = frozenset(["completed", "failed", "exception"])

    def __call__(self, git_gecko: Repo, git_wpt: Repo, body: Dict[str, Any]) -> None:
        newrelic.agent.set_transaction_name("DecisionTaskHandler")
        task_id = body["status"]["taskId"]
        taskgroup_id = body["status"]["taskGroupId"]

        msg = "Expected kind decision-task, got %s" % body["task"]["tags"]["kind"]
        assert body["task"]["tags"]["kind"] == "decision-task", msg

        newrelic.agent.add_custom_parameter("tc_task", task_id)
        newrelic.agent.add_custom_parameter("tc_taskgroup", taskgroup_id)

        state = body["status"]["state"]
        newrelic.agent.add_custom_parameter("state", state)

        # Enforce the invariant that the taskgroup id is not set until
        # the decision task is complete. This allows us to determine if a
        # try push should have the expected wpt tasks just by checking if
        # this is set
        if state not in self.complete_states:
            logger.info("Decision task is not yet complete, status %s" % state)
            return

        task = tc.get_task(task_id)
        if task is None:
            raise ValueError("Failed to get task for task_id %s" % task_id)

        sha1 = task.get("payload", {}).get("env", {}).get("GECKO_HEAD_REV")

        if sha1 is None:
            raise ValueError("Failed to get commit sha1 from task message")

        if state == "exception":
            run_id = body["runId"]
            runs = body.get("status", {}).get("runs", [])
            if 0 <= run_id < len(runs):
                reason = runs[run_id].get("reasonResolved")
                if reason in ["superseded",
                              "claim-expired",
                              "worker-shutdown",
                              "intermittent-task"]:
                    logger.info("Task %s had an exception for reason %s, "
                                "assuming taskcluster will retry" %
                                (task_id, reason))
                    return

        try_push = trypush.TryPush.for_commit(git_gecko, sha1)
        if not try_push:
            logger.debug(f"No try push for SHA1 {sha1} taskId {task_id}")
            # This could be a race condition if the decision task completes before this
            # task is in the index
            raise RetryableError("Got a wptsync task with no corresponding try push")

        with SyncLock.for_process(try_push.process_name) as lock:
            assert isinstance(lock, SyncLock)
            with try_push.as_mut(lock):
                # If we retrigger, we create a new taskgroup, with id equal to the new task_id.
                # But the retriggered decision task itself is still in the original taskgroup
                if state == "completed":
                    logger.info("Setting taskgroup id for try push %r to %s" %
                                (try_push, taskgroup_id))
                    try_push.taskgroup_id = taskgroup_id
                elif state in ("failed", "exception"):
                    sync = try_push.sync(git_gecko, git_wpt)
                    message = ("Decision task got status %s for task %s%s" %
                               (state, sha1, " PR %s" % sync.pr if sync and sync.pr else ""))
                    logger.error(message)
                    taskgroup = tc.TaskGroup(task["taskGroupId"])
                    if len(taskgroup.view(
                            lambda x: x["task"]["metadata"]["name"] == "Gecko Decision Task")) > 5:
                        try_push.status = "complete"
                        try_push.infra_fail = True
                        try_push.taskgroup_id = taskgroup_id
                        if sync and sync.bug:
                            env.bz.comment(
                                sync.bug,
                                "Try push failed: decision task %s returned error" % task_id)
                    else:
                        logger.info("Retriggering decision task for sync %s" %
                                    (sync.process_name,))
                        client = tc.TaskclusterClient()
                        client.retrigger(task_id)


class TryTaskHandler(Handler):
    """Handler for the task associated with a try push task completing."""

    def __call__(self, git_gecko: Repo, git_wpt: Repo, body: Dict[str, Any]) -> None:
        newrelic.agent.set_transaction_name("TryTaskHandler")
        taskgroup_id = body["status"]["taskGroupId"]
        newrelic.agent.add_custom_parameter("tc_taskgroup", taskgroup_id)

        try_push = trypush.TryPush.for_taskgroup(git_gecko, taskgroup_id)
        if not try_push:
            logger.debug("No try push for taskgroup %s" % taskgroup_id)
            # this is not one of our try_pushes
            return

        if try_push.status == "complete":
            return

        logger.info("Found try push for taskgroup %s" % taskgroup_id)

        # Check if the taskgroup has all tasks complete, excluding unscheduled tasks.
        # This allows us to tell if the taskgroup is complete (per the treeherder definition)
        # even when there are tasks that won't be scheduled because the task they depend on
        # failed. Otherwise we'd have to wait until those unscheduled tasks time out, which
        # usually takes 24hr
        if try_push.taskgroup_id is None:
            with SyncLock.for_process(try_push.process_name) as lock:
                assert isinstance(lock, SyncLock)
                with try_push.as_mut(lock):
                    try_push.taskgroup_id = taskgroup_id
        tasks = try_push.tasks()
        if tasks.complete(allow_unscheduled=True):
            taskgroup_complete(git_gecko, git_wpt, taskgroup_id, try_push)


class TaskGroupHandler(Handler):
    def __call__(self, git_gecko: Repo, git_wpt: Repo, body: Dict[str, Any]) -> None:
        newrelic.agent.set_transaction_name("TaskGroupHandler")
        taskgroup_id = tc.normalize_task_id(body["taskGroupId"])

        newrelic.agent.add_custom_parameter("tc_task", taskgroup_id)

        try_push = trypush.TryPush.for_taskgroup(git_gecko, taskgroup_id)
        if not try_push:
            logger.debug("No try push for taskgroup %s" % taskgroup_id)
            # this is not one of our try_pushes
            return
        logger.info("Found try push for taskgroup %s" % taskgroup_id)
        taskgroup_complete(git_gecko, git_wpt, taskgroup_id, try_push)


def taskgroup_complete(git_gecko: Repo, git_wpt: Repo, taskgroup_id: str,
                       try_push: trypush.TryPush) -> None:
    sync = try_push.sync(git_gecko, git_wpt)
    if not sync:
        newrelic.agent.record_custom_event("taskgroup_sync_missing", params={
            "taskgroup-id": taskgroup_id,
            "try_push": try_push,
        })
        return

    with SyncLock.for_process(sync.process_name) as lock:
        assert isinstance(lock, SyncLock)
        with sync.as_mut(lock), try_push.as_mut(lock):
            # We sometimes see the taskgroup ID being None. If it isn't set but found via its
            # taskgroup ID, it is safe to set it here.
            if try_push.taskgroup_id is None:
                logger.info("Try push for taskgroup %s does not have its ID set, setting now" %
                            taskgroup_id)
                try_push.taskgroup_id = taskgroup_id  # type: ignore
                newrelic.agent.record_custom_event("taskgroup_id_missing", params={
                    "taskgroup-id": taskgroup_id,
                    "try_push": try_push,
                    "sync": sync,
                })
            elif try_push.taskgroup_id != taskgroup_id:
                msg = ("TryPush %s, expected taskgroup ID %s, found %s instead" %
                       (try_push, taskgroup_id, try_push.taskgroup_id))
                logger.error(msg)
                exc = ValueError(msg)
                newrelic.agent.record_exception(exc=exc)
                raise exc

            if sync:
                logger.info("Updating try push for sync %r" % sync)
            if isinstance(sync, downstream.DownstreamSync):
                downstream.try_push_complete(git_gecko, git_wpt, try_push, sync)
            elif isinstance(sync, landing.LandingSync):
                landing.try_push_complete(git_gecko, git_wpt, try_push, sync)


class LandingHandler(Handler):
    def __call__(self, git_gecko: Repo, git_wpt: Repo, body: Dict[str, Any]) -> None:
        newrelic.agent.set_transaction_name("LandingHandler")
        landing.update_landing(git_gecko, git_wpt)


class CleanupHandler(Handler):
    def __call__(self, git_gecko: Repo, git_wpt: Repo, body: Dict[str, Any]) -> None:
        newrelic.agent.set_transaction_name("CleanupHandler")
        logger.info("Running cleanup")
        worktree.cleanup(git_gecko, git_wpt)
        tc.cleanup()


class RetriggerHandler(Handler):
    def __call__(self, git_gecko: Repo, git_wpt: Repo, body: Dict[str, Any]) -> None:
        newrelic.agent.set_transaction_name("RetriggerHandler")
        logger.info("Running retrigger")
        update_repositories(git_gecko, git_wpt)
        sync_point = landing.load_sync_point(git_gecko, git_wpt)
        prev_wpt_head = sync_point["upstream"]
        unlanded = landing.unlanded_with_type(git_gecko, git_wpt, None, prev_wpt_head)
        update.retrigger(git_gecko, git_wpt, unlanded)


class PhabricatorHandler(Handler):
    def __call__(self, git_gecko: Repo, git_wpt: Repo, body: Dict[str, Any]) -> None:
        newrelic.agent.set_transaction_name("PhabricatorHandler")
        logger.info('Got phab event, doing nothing: %s' % body)


class BugUpdateHandler(Handler):
    def __call__(self, git_gecko: Repo, git_wpt: Repo, body: Dict[str, Any]) -> None:
        newrelic.agent.set_transaction_name("BugUpdateHandler")
        logger.info("Running bug update")
        bugupdate.update_triage_bugs(git_gecko)