bot/code_review_bot/workflow.py

# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.

import asyncio
import time
from datetime import datetime, timedelta
from itertools import groupby

import structlog
from libmozdata.phabricator import BuildState, PhabricatorAPI
from libmozevent.mercurial import MercurialWorker, Repository
from libmozevent.phabricator import PhabricatorActions, PhabricatorBuildState
from taskcluster.utils import stringDate

from code_review_bot import Level, stats
from code_review_bot.analysis import (
    RevisionBuild,
    publish_analysis_lando,
    publish_analysis_phabricator,
)
from code_review_bot.backend import BackendAPI
from code_review_bot.config import REPO_AUTOLAND, REPO_MOZILLA_CENTRAL, settings
from code_review_bot.mercurial import robust_checkout
from code_review_bot.report.debug import DebugReporter
from code_review_bot.revisions import Revision
from code_review_bot.tasks.base import AnalysisTask, BaseTask, NoticeTask
from code_review_bot.tasks.clang_format import ClangFormatTask
from code_review_bot.tasks.clang_tidy import ClangTidyTask
from code_review_bot.tasks.clang_tidy_external import ExternalTidyTask
from code_review_bot.tasks.coverage import ZeroCoverageTask
from code_review_bot.tasks.default import DefaultTask
from code_review_bot.tasks.docupload import DocUploadTask
from code_review_bot.tasks.lint import MozLintTask
from code_review_bot.tasks.tgdiff import TaskGraphDiffTask

logger = structlog.get_logger(__name__)

TASKCLUSTER_NAMESPACE = "project.relman.{channel}.code-review.{name}"
TASKCLUSTER_INDEX_TTL = 7  # in days


class Workflow:
    """
    Full static analysis workflow
    - setup remote analysis workflow
    - find issues from remote tasks
    - publish issues
    """

    def __init__(
        self,
        reporters,
        index_service,
        queue_service,
        phabricator_api,
        zero_coverage_enabled=True,
        update_build=True,
        task_failures_ignored=[],
    ):
        self.zero_coverage_enabled = zero_coverage_enabled
        self.update_build = update_build
        self.task_failures_ignored = task_failures_ignored
        logger.info("Will ignore task failures", names=self.task_failures_ignored)

        # Use share phabricator API client
        assert isinstance(phabricator_api, PhabricatorAPI)
        self.phabricator = phabricator_api

        # Load reporters to use
        self.reporters = reporters
        if not self.reporters:
            logger.warn("No reporters configured, this analysis will not be published")

        # Always add debug reporter and Diff reporter
        self.reporters["debug"] = DebugReporter(
            output_dir=settings.taskcluster.results_dir
        )

        # Use TC services client
        self.index_service = index_service
        self.queue_service = queue_service

        # Setup Backend API client
        self.backend_api = BackendAPI()

        # Is local clone already setup ?
        self.clone_available = False

    def run(self, revision):
        """
        Find all issues on remote tasks and publish them
        """
        # Index ASAP Taskcluster task for this revision
        self.index(revision, state="started")

        # Set the Phabricator build as running
        self.update_status(revision, state=BuildState.Work)
        if settings.taskcluster_url:
            self.publish_link(
                revision,
                slug="publication",
                name="Publication task",
                url=settings.taskcluster_url,
            )

        # Analyze revision patch to get files/lines data
        revision.analyze_patch()

        # Find issues on remote tasks
        issues, task_failures, notices, reviewers = self.find_issues(
            revision, settings.try_group_id
        )

        # Analyze issues in case the before/after feature is enabled
        if revision.before_after_feature:
            logger.info("Running the before/after feature")
            # Search a base revision from the decision task
            decision = self.queue_service.task(settings.try_group_id)
            base_rev_changeset = (
                decision.get("payload", {}).get("env", {}).get("GECKO_BASE_REV")
            )
            if not base_rev_changeset:
                logger.warning(
                    "Base revision changeset could not be fetched from Phabricator, "
                    "looking for existing issues based on the current date",
                    task=settings.try_group_id,
                )

            # Clone local repo when required
            # as find_previous_issues will build the hashes
            self.clone_repository(revision)

            # Mark know issues to avoid publishing them on this patch
            self.find_previous_issues(issues, base_rev_changeset)
            new_issues_count = sum(issue.new_issue for issue in issues)
            logger.info(
                f"Found {new_issues_count} new issues (over {len(issues)} total detected issues)",
                task=settings.try_group_id,
            )
        else:
            # Clone local repo when required
            # as publication need the hashes
            self.clone_repository(revision)

        if (
            all(issue.new_issue is False for issue in issues)
            and not task_failures
            and not notices
        ):
            logger.info("No issues nor notices, stopping there.")

        # Publish all issues
        self.publish(revision, issues, task_failures, notices, reviewers)

        return issues

    def ingest_revision(self, revision, group_id):
        """
        Simpler workflow to ingest a revision
        """
        assert revision.head_repository in (
            REPO_AUTOLAND,
            REPO_MOZILLA_CENTRAL,
        ), "Need a revision from autoland or mozilla-central"
        logger.info(
            "Starting revision ingestion",
            bugzilla=revision.bugzilla_id,
            title=revision.title,
            head_repository=revision.head_repository,
            head_changeset=revision.head_changeset,
        )

        assert (
            self.backend_api.enabled
        ), "Backend storage is disabled, revision ingestion is not possible"

        supported_tasks = []

        def _build_tasks(tasks):
            for task_status in tasks["tasks"]:
                try:
                    task_name = task_status["task"]["metadata"]["name"]
                    # Only analyze tasks stating with `source-test-` to avoid checking artifacts every time
                    if not task_name.startswith("source-test-"):
                        logger.debug(
                            f"Task with name '{task_name}' is not supported during the ingestion of a revision"
                        )
                        continue
                    task = self.build_task(task_status)
                except Exception as e:
                    logger.warning(f"Could not proceed task {task_name}: {e}")
                    continue
                if task is None or getattr(task, "parse_issues", None) is None:
                    # Do ignore tasks that cannot be parsed as issues
                    continue
                supported_tasks.append(task)

        # Find potential issues in the task group
        self.queue_service.listTaskGroup(group_id, paginationHandler=_build_tasks)
        logger.info(
            "Loaded all supported tasks in the task group",
            group_id=group_id,
            nb=len(supported_tasks),
        )

        # Load all the artifacts and potential issues
        issues = []
        for task in supported_tasks:
            artifacts = task.load_artifacts(self.queue_service)
            if artifacts is not None:
                task_issues = task.parse_issues(artifacts, revision)
                logger.info(
                    f"Found {len(task_issues)} issues",
                    task=task.name,
                    id=task.id,
                )
                issues += task_issues

        # Store the revision & diff in the backend
        self.backend_api.publish_revision(revision)

        # Publish issues when there are some
        if not issues:
            logger.info("No issues for that revision")
            return

        # Clone local repo when required
        self.clone_repository(revision)

        # Publish issues in the backend
        self.backend_api.publish_issues(issues, revision)

    def start_analysis(self, revision):
        """
        Apply a patch on a local clone and push to try to trigger a new Code review analysis
        """
        logger.info("Starting revision analysis", revision=revision)

        # Do not process revisions from black-listed users
        if revision.is_blacklisted:
            logger.warning("Blacklisted author, stopping there.")
            return

        # Cannot run without mercurial cache configured
        if not settings.mercurial_cache:
            raise Exception("Mercurial cache must be configured to start analysis")

        # Cannot run without ssh key
        if not settings.ssh_key:
            raise Exception("SSH Key must be configured to start analysis")

        # Set the Phabricator build as running
        self.update_status(revision, state=BuildState.Work)
        if settings.taskcluster_url:
            self.publish_link(
                revision,
                slug="analysis",
                name="Analysis task",
                url=settings.taskcluster_url,
            )

        # Initialize Phabricator build using revision
        build = RevisionBuild(revision)

        # Copy internal Phabricator credentials to setup libmozevent
        phabricator = PhabricatorActions(
            url=self.phabricator.url,
            api_key=self.phabricator.api_key,
        )

        # Initialize mercurial repository
        repository = Repository(
            config={
                "name": revision.base_repository_conf.name,
                "try_name": revision.base_repository_conf.try_name,
                "url": revision.base_repository_conf.url,
                "try_url": revision.base_repository_conf.try_url,
                # Setup ssh identity
                "ssh_user": revision.base_repository_conf.ssh_user,
                "ssh_key": settings.ssh_key,
                # Force usage of robustcheckout
                "checkout": "robust",
            },
            cache_root=settings.mercurial_cache,
        )

        worker = MercurialWorker(
            # We are not using the mercurial workflow
            # so we can initialize with empty data here
            queue_name=None,
            queue_phabricator=None,
            repositories={},
        )

        # Try to update the state 5 consecutive time
        for i in range(5):
            # Update the internal build state using Phabricator infos
            phabricator.update_state(build)

            # Continue with workflow once the build is public
            if build.state is PhabricatorBuildState.Public:
                break

            # Retry later if the build is not yet seen as public
            logger.warning(
                "Build is not public, retrying in 30s",
                build=build,
                retries_left=build.retries,
            )
            time.sleep(30)

        # Make sure the build is now public
        if build.state is not PhabricatorBuildState.Public:
            raise Exception("Cannot process private builds")

        # When the build is public, load needed details
        try:
            phabricator.load_patches_stack(build)
            logger.info("Loaded stack of patches", build=str(build))
        except Exception as e:
            logger.warning(
                "Failed to load build details", build=str(build), error=str(e)
            )
            raise

        if not build.stack:
            raise Exception("No stack of patches to apply.")

        # We'll clone the required repository
        repository.clone()

        # Apply the stack of patches using asynchronous method
        # that runs directly in that process
        output = asyncio.run(worker.handle_build(repository, build))

        # Update final state using worker output
        if self.update_build:
            publish_analysis_phabricator(output, self.phabricator)
        else:
            logger.info("Skipping Phabricator publication")

        # Send Build in progress or errors to Lando
        lando_reporter = self.reporters.get("lando")
        if lando_reporter is not None:
            publish_analysis_lando(output, lando_reporter.lando_api)
        else:
            logger.info("Skipping Lando publication")

    def clone_repository(self, revision):
        """
        Clone the repo locally when configured
        On production this should use a Taskcluster cache
        """
        if not settings.mercurial_cache:
            logger.debug("Local clone not required")
            return
        if self.clone_available:
            logger.debug("Local clone already setup")
            return

        logger.info(
            "Cloning revision to build issues",
            repo=revision.base_repository,
            changeset=revision.head_changeset,
            dest=settings.mercurial_cache_checkout,
        )
        robust_checkout(
            repo_upstream_url=revision.base_repository,
            repo_url=revision.head_repository,
            revision=revision.head_changeset,
            checkout_dir=settings.mercurial_cache_checkout,
            sharebase_dir=settings.mercurial_cache_sharebase,
        )
        self.clone_available = True

    def publish(self, revision, issues, task_failures, notices, reviewers):
        """
        Publish issues on selected reporters
        """
        # Publish patches on Taskcluster
        # or write locally for local development
        for patch in revision.improvement_patches:
            if settings.taskcluster.local:
                patch.write()
            else:
                patch.publish()

        # Publish issues on backend to retrieve their comparison state
        publishable_issues = [i for i in issues if i.is_publishable()]

        self.backend_api.publish_issues(publishable_issues, revision)

        # Report issues publication stats
        nb_issues = len(issues)
        nb_publishable = len(publishable_issues)
        nb_publishable_errors = len(
            [i for i in publishable_issues if i.level == Level.Error]
        )

        self.index(
            revision,
            state="analyzed",
            issues=nb_issues,
            issues_publishable=nb_publishable,
        )
        stats.add_metric("analysis.issues.publishable", nb_publishable)

        # Publish reports about these issues
        with stats.timer("runtime.reports"):
            for reporter in self.reporters.values():
                reporter.publish(issues, revision, task_failures, notices, reviewers)

        self.index(
            revision, state="done", issues=nb_issues, issues_publishable=nb_publishable
        )

        # Publish final HarborMaster state
        self.update_status(
            revision,
            BuildState.Fail
            if nb_publishable_errors > 0 or task_failures
            else BuildState.Pass,
        )

    def index(self, revision, **kwargs):
        """
        Index current task on Taskcluster index
        """
        assert isinstance(revision, Revision)

        if settings.taskcluster.local or self.index_service is None:
            logger.info("Skipping taskcluster indexing", rev=str(revision), **kwargs)
            return

        # Build payload
        payload = revision.as_dict()
        payload.update(kwargs)

        # Always add the indexing
        now = datetime.utcnow()
        payload["indexed"] = stringDate(now)

        # Always add the source and try config
        payload["source"] = "try"
        payload["try_task_id"] = settings.try_task_id
        payload["try_group_id"] = settings.try_group_id

        # Always add the repository we are working on
        # This is mainly used by the frontend to list & filter diffs
        payload["repository"] = revision.base_repository

        # Add restartable flag for monitoring
        payload["monitoring_restart"] = payload["state"] == "error" and payload.get(
            "error_code"
        ) in ("watchdog", "mercurial")

        # Add a sub namespace with the task id to be able to list
        # tasks from the parent namespace
        namespaces = revision.namespaces + [
            f"{namespace}.{settings.taskcluster.task_id}"
            for namespace in revision.namespaces
        ]

        # Build complete namespaces list, with monitoring update
        full_namespaces = [
            TASKCLUSTER_NAMESPACE.format(channel=settings.app_channel, name=name)
            for name in namespaces
        ]

        # Index for all required namespaces
        for namespace in full_namespaces:
            self.index_service.insertTask(
                namespace,
                {
                    "taskId": settings.taskcluster.task_id,
                    "rank": 0,
                    "data": payload,
                    "expires": stringDate(now + timedelta(days=TASKCLUSTER_INDEX_TTL)),
                },
            )

    def find_previous_issues(self, issues, base_rev_changeset=None):
        """
        Look for known issues in the backend matching the given list of issues

        If a base revision ID is provided, compare to issues detected on this revision
        Otherwise, compare to issues detected on last ingested revision
        """
        assert (
            self.backend_api.enabled
        ), "Backend storage is disabled, comparing issues is not possible"

        current_date = datetime.now().strftime("%Y-%m-%d")

        # Group issues by path, so we only list know issues for the affected files
        issues_groups = groupby(
            sorted(issues, key=lambda i: i.path),
            lambda i: i.path,
        )
        logger.info(
            "Checking for existing issues in the backend",
            base_revision_changeset=base_rev_changeset,
        )

        for path, group_issues in issues_groups:
            known_issues = self.backend_api.list_repo_issues(
                "mozilla-central",
                date=current_date,
                revision_changeset=base_rev_changeset,
                path=path,
            )
            hashes = [issue["hash"] for issue in known_issues]
            for issue in group_issues:
                issue.new_issue = bool(issue.hash and issue.hash not in hashes)

    def find_issues(self, revision, group_id):
        """
        Find all issues on remote Taskcluster task group
        """
        # Load all tasks in task group
        tasks = self.queue_service.listTaskGroup(group_id)
        assert "tasks" in tasks
        tasks = {task["status"]["taskId"]: task for task in tasks["tasks"]}
        assert len(tasks) > 0
        logger.info("Loaded Taskcluster group", id=group_id, tasks=len(tasks))

        # Store the revision in the backend (or retrieve an existing one)
        rev = self.backend_api.publish_revision(revision)
        assert (
            rev is not None
        ), "Stopping early because revision could not be created nor retrieved from the backend"

        # Load task description
        task = tasks.get(settings.try_task_id)
        assert task is not None, f"Missing task {settings.try_task_id}"
        dependencies = task["task"]["dependencies"]
        assert len(dependencies) > 0, "No task dependencies to analyze"

        # Skip dependencies not in group
        # But log all skipped tasks
        def _in_group(dep_id):
            if dep_id not in tasks:
                # Used for docker images produced in tree
                # and other artifacts
                logger.info("Skip dependency not in group", task_id=dep_id)
                return False
            return True

        dependencies = [dep_id for dep_id in dependencies if _in_group(dep_id)]

        # Do not run parsers when we only have a gecko decision task
        # That means no analyzer were triggered by the taskgraph decision task
        # This can happen if the patch only touches file types for which we have no analyzer defined
        # See issue https://github.com/mozilla/release-services/issues/2055
        if len(dependencies) == 1:
            task = tasks[dependencies[0]]
            if task["task"]["metadata"]["name"] == "Gecko Decision Task":
                logger.warn("Only dependency is a Decision Task, skipping analysis")
                return [], [], [], []

        # Add zero-coverage task
        if self.zero_coverage_enabled:
            dependencies.append(ZeroCoverageTask)

        # Find issues and patches in dependencies
        issues = []
        task_failures = []
        notices = []
        for dep in dependencies:
            try:
                if isinstance(dep, type) and issubclass(dep, AnalysisTask):
                    # Build a class instance from its definition and route
                    task = dep.build_from_route(self.index_service, self.queue_service)
                else:
                    # Use a task from its id & description
                    task = self.build_task(tasks[dep])
                if task is None:
                    continue
                artifacts = task.load_artifacts(self.queue_service)
                if artifacts is not None:
                    task_issues, task_patches = [], []
                    if isinstance(task, AnalysisTask):
                        task_issues = task.parse_issues(artifacts, revision)
                        logger.info(
                            f"Found {len(task_issues)} issues",
                            task=task.name,
                            id=task.id,
                        )
                        stats.report_task(task, task_issues)
                        issues += task_issues

                        task_patches = task.build_patches(artifacts)
                        for patch in task_patches:
                            revision.add_improvement_patch(task, patch)

                    elif isinstance(task, NoticeTask):
                        notice = task.build_notice(artifacts, revision)
                        if notice:
                            notices.append(notice)

                    # Report a problem when tasks in erroneous state are found
                    # but no issue or patch has been processed by the bot
                    if task.state == "failed" and not task_issues and not task_patches:
                        # Skip task that are listed as ignorable (we try to avoid unnecessary spam)
                        if task.name in self.task_failures_ignored:
                            logger.warning(
                                "Ignoring task failure as configured",
                                task=task.name,
                                id=task.id,
                            )
                            continue

                        logger.warning(
                            "An erroneous task processed some artifacts and found no issues or patches",
                            task=task.name,
                            id=task.id,
                        )
                        task_failures.append(task)
            except Exception as e:
                logger.warn(
                    "Failure during task analysis",
                    task=settings.taskcluster.task_id,
                    error=e,
                )
                raise

        reviewers = (
            task.extra_reviewers_groups if task and isinstance(task, BaseTask) else []
        )
        return issues, task_failures, notices, reviewers

    def build_task(self, task_status):
        """
        Create a specific implementation of AnalysisTask according to the task name
        """
        try:
            task_id = task_status["status"]["taskId"]
        except KeyError:
            raise Exception(f"Cannot read task name {task_id}")
        try:
            name = task_status["task"]["metadata"]["name"]
        except KeyError:
            raise Exception(f"Cannot read task name {task_id}")

        # Default format is used first when the correct artifact is available
        if DefaultTask.matches(task_id):
            return DefaultTask(task_id, task_status)
        elif name.startswith("source-test-mozlint-"):
            return MozLintTask(task_id, task_status)
        elif name == "source-test-clang-tidy":
            return ClangTidyTask(task_id, task_status)
        elif name == "source-test-clang-format":
            return ClangFormatTask(task_id, task_status)
        elif name == "source-test-doc-upload":
            return DocUploadTask(task_id, task_status)
        elif name == "source-test-clang-external":
            return ExternalTidyTask(task_id, task_status)
        elif name == "source-test-taskgraph-diff":
            return TaskGraphDiffTask(task_id, task_status)
        elif settings.autoland_group_id is not None and not name.startswith(
            "source-test-"
        ):
            # Log cleanly on autoland unknown tasks
            logger.info("Skipping unknown task", id=task_id, name=name)
        else:
            return None

    def update_status(self, revision, state):
        """
        Update build status on HarborMaster
        """
        assert isinstance(state, BuildState)
        if not revision.build_target_phid:
            logger.info(
                "No build target found, skipping HarborMaster update", state=state.value
            )
            return

        if not self.update_build:
            logger.info(
                "Update build disabled, skipping HarborMaster update", state=state.value
            )
            return

        self.phabricator.update_build_target(revision.build_target_phid, state)
        logger.info("Updated HarborMaster status", state=state, revision=revision)

    def publish_link(self, revision: Revision, slug: str, name: str, url: str):
        """
        Publish a link as a HarborMaster artifact
        """
        if not revision.build_target_phid:
            logger.info(
                "No build target found, skipping HarborMaster link creation",
                slug=slug,
                url=url,
            )
            return

        if not self.update_build:
            logger.info(
                "Update build disabled, skipping HarborMaster link creation",
                slug=slug,
                url=url,
            )
            return

        self.phabricator.create_harbormaster_uri(
            revision.build_target_phid, slug, name, url
        )