recipe_modules/autocorrelator/api.py

# Copyright 2021 The Fuchsia Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

from recipe_engine import recipe_api

AUTOCORRELATOR_PATH = "fuchsia/infra/autocorrelator/${platform}"
AUTOCORRELATOR_VERSION = "git_revision:25ebd0de9766d0a16ca0fdf39618576e7b597369"


class AutocorrelatorApi(recipe_api.RecipeApi):
    """APIs for finding correlations between failures in Fuchsia CI/CQ."""

    def __init__(self, *args, **kwargs):
        super(AutocorrelatorApi, self).__init__(*args, **kwargs)
        self.findings = []

    def check_ci(
        self,
        step_name,
        base_commit,
        builder,
        build_status,
        summary_markdown,
        score_threshold=0.95,
    ):
        """Compare summary markdown similarity to a CI-like builder. Record a
        finding if it meets the score threshold.

        Args:
            step_name (str): Name of the step.
            base_commit (str): Base commit as sha1 to check against `builder`.
            builder (str): Fully-qualified Buildbucket builder name of the
                CI-like builder to check.
            build_status (common_pb2.Status): Buildbucket status.
            summary_markdown (str): Summary markdown to compare.
            score_threshold (float): Record a finding if it meets this
                threshold.
        """
        with self.m.step.nest(step_name):
            args = [
                "check-ci",
                "-base-commit",
                base_commit,
                "-builder",
                builder,
                "-build-status",
                build_status,
                "-summary-markdown-path",
                self._summary_markdown_path(summary_markdown),
                "-json-output",
                self.m.json.output(),
            ]
            step = self("run autocorrelator", args)
            finding = step.json.output
            if not finding or finding["is_green"]:
                return step
            score = finding["score"]
            if score >= score_threshold:
                ci_build_link = self.m.buildbucket.build_url(
                    build_id=finding["build_id"]
                )
                step.presentation.links["correlated_ci_build"] = ci_build_link
                self.findings.append(
                    "Correlated failure found in CI %s: %0.2f similarity, %d git "
                    "commit distance" % (ci_build_link, score, finding["commit_dist"]),
                )
            return step

    def check_try(
        self,
        step_name,
        builder,
        change_num,
        build_status,
        summary_markdown,
        ignore_skipped_build=False,
        ignore_skipped_tests=False,
        score_threshold=0.95,
        build_frequency_threshold=0.75,
    ):
        """Compare summary markdown similarity to a try-like builder. Record an
        aggregate finding if there are one or more findings that meet the score
        threshold, and the frequency of findings meets the build frequency
        threshold.

        Args:
            step_name (str): Name of the step.
            builder (str): Fully-qualified Buildbucket builder name of the
                try-like builder to check.
            change_num (int): Gerrit change number.
            build_status (common_pb2.Status): Buildbucket status.
            summary_markdown (str): Summary markdown to compare.
            ignore_skipped_build (bool): Whether to ignore try builds with
                unaffected build graphs.
            ignore_skipped_tests (bool): Whether to ignore builds which skipped
                testing.
            score_threshold (float): Count a finding as part of the aggregate if
                its score meets this threshold.
            build_frequency_threshold (float): Record an aggregate finding if
                the frequency of findings meets this threshold, i.e. a 0.5
                threshold means that a finding will be recorded if
                num_findings / num_inspected_builds >= 0.5.
        """
        with self.m.step.nest(step_name):
            args = [
                "check-try",
                "-builder",
                builder,
                "-change-num",
                change_num,
                "-build-status",
                build_status,
                "-summary-markdown-path",
                self._summary_markdown_path(summary_markdown),
                "-json-output",
                self.m.json.output(),
            ]
            if ignore_skipped_build:
                args.append("-ignore-skipped-build")
            if ignore_skipped_tests:
                args.append("-ignore-skipped-tests")
            step = self("run autocorrelator", args)
            findings = step.json.output
            if not findings:
                return step
            findings_meeting_threshold = []
            for finding in findings:
                if finding["score"] >= score_threshold:
                    findings_meeting_threshold.append(finding)
                    try_build_link = self.m.buildbucket.build_url(
                        build_id=finding["build_id"]
                    )
                    step.presentation.links[
                        "correlated_try_build: %s" % finding["build_id"]
                    ] = try_build_link
            build_frequency = len(findings_meeting_threshold) / float(len(findings))
            if build_frequency >= build_frequency_threshold:
                self.findings.append(
                    "Correlated failures found in try: %0.2f avg similarity in %d "
                    "builds\n\n%s"
                    % (
                        sum(
                            (finding["score"] for finding in findings_meeting_threshold)
                        )
                        / len(findings_meeting_threshold),
                        len(findings_meeting_threshold),
                        "\n".join(
                            build_link
                            for build_link in step.presentation.links.itervalues()
                        ),
                    ),
                )
            return step

    def __call__(self, step_name, args):
        return self.m.step(
            step_name, [self._autocorrelator_tool] + args, infra_step=True
        )

    def _summary_markdown_path(self, summary_markdown):
        """Write summary markdown to a file and return the Path."""
        path = self.m.path.mkstemp("summary_md")
        self.m.file.write_text("write summary markdown", path, summary_markdown)
        return path

    @property
    def _autocorrelator_tool(self):
        return self.m.cipd.ensure_tool(AUTOCORRELATOR_PATH, AUTOCORRELATOR_VERSION)