recipe_modules/display_util/api.py

# Copyright 2020 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

from PB.go.chromium.org.luci.buildbucket.proto import common as common_pb2

from RECIPE_MODULES.fuchsia.utils import pluralize


class DisplayUtilApi(recipe_api.RecipeApi):
    """Module to display buildbucket builds or swarming tasks as steps."""

    def display_builds(self, step_name, builds, raise_on_failure=False):
        """Display build links and status for each input build.

        Optionally raise on build failure(s).

        Args:
            step_name (str): Name of build group to display in step name.
            builds (seq(buildbucket.v2.Build)): buildbucket Build objects.
                See recipe_engine/buildbucket recipe module for more info.
            raise_on_failure (bool): Raise InfraFailure or StepFailure on
                failure.

        Raises:
            InfraFailure: One or more input builds had infra failure. Takes
                priority over step failures.
            StepFailure: One or more of input builds failed.
        """
        builds = list(builds)

        failures = []
        infra_failures = []

        with self.m.step.nest(step_name) as presentation:
            for b in builds:
                with self.m.step.nest(b.builder.builder) as display_step:
                    display_step.presentation.links[
                        str(b.id)
                    ] = self.m.buildbucket.build_url(build_id=b.id)
                    if b.status == common_pb2.INFRA_FAILURE:
                        display_step.presentation.status = self.m.step.EXCEPTION
                        infra_failures.append(b)
                    elif b.status == common_pb2.FAILURE:
                        display_step.presentation.status = self.m.step.FAILURE
                        failures.append(b)
                    else:
                        if b.status != common_pb2.SUCCESS:
                            # For any other status, use warning color.
                            display_step.presentation.status = self.m.step.WARNING
                        continue

            num_failed = len(failures) + len(infra_failures)
            if num_failed == 0:
                return

            failure_message_parts = []

            # If the build failed, include it in the final message, truncating
            # its summary_markdown. This is a very conservative estimate about
            # how much of each build's summary we should truncate. May not work
            # well with large numbers of failed builds.
            truncate_length = self.m.buildbucket_util.MAX_SUMMARY_SIZE / (
                2 * num_failed
            )
            for b in infra_failures + failures:
                failure_message_parts.append(self._summary_section(b, truncate_length))

            if not raise_on_failure:  # pragma: no cover
                return

            # If there were any infra failures, raise purple.
            if infra_failures:
                presentation.status = self.m.step.EXCEPTION
                exception_type = self.m.step.InfraFailure
            # Otherwise if there were any step failures, raise red.
            else:
                presentation.status = self.m.step.FAILURE
                exception_type = self.m.step.StepFailure

        raise exception_type(
            "%s failed:\n\n%s"
            % (pluralize("build", num_failed), "\n\n".join(failure_message_parts))
        )

    def _summary_section(self, build, truncate_length):
        url = self.m.buildbucket.build_url(build_id=build.id)
        failure_header = "[%s](%s)" % (build.builder.builder, url)
        if build.status == common_pb2.INFRA_FAILURE:
            failure_header += " (infra failure)"

        summary = self.m.buildbucket_util.summary_message(
            build.summary_markdown.strip(),
            truncation_message="(truncated)",
            escape_markdown=False,
            truncate_length=truncate_length,
        )
        # Don't include an empty summary, or any summaries if we'll
        # have to truncate them by so much that they won't contain much
        # useful information.
        if truncate_length < 75 or not summary:
            return failure_header
        return failure_header + ":\n\n%s" % summary

    def display_tasks(self, step_name, results, metadata, raise_on_failure=False):
        """Display task links and status for each input task.

        Optionally raise on task failure(s).

        Args:
          step_name (str): Name of build group to display in step name.
          results (seq(swarming.TaskResult)): swarming TaskResult objects. See
            recipe_engine/swarming recipe module for more info.
          metadata (seq(swarming.TaskMetadata)): swarming TaskMetadata objects. See
            recipe_engine/swarming recipe module for more info.
          raise_on_failure (bool): Raise InfraFailure or StepFailure on failure.

        Raises:
          InfraFailure: One or more input tasks had infra failure. Takes priority
            over step failures.
          StepFailure: One or more of input tasks failed.
        """
        infra_failed_tasks = []
        failed_tasks = []
        links = {m.id: m.task_ui_link for m in metadata}
        with self.m.step.nest(step_name) as presentation:
            for result in results:
                with self.m.step.nest(result.name) as display_step:
                    step_links = display_step.presentation.links
                    link = links[result.id]
                    step_links[str(result.id)] = link
                    if (
                        result.state is None
                        or result.state != self.m.swarming.TaskState.COMPLETED
                    ):
                        display_step.status = self.m.step.EXCEPTION
                        infra_failed_tasks.append(result.name)
                    elif not result.success:
                        display_step.status = self.m.step.FAILURE
                        failed_tasks.append(result.name)
                    else:
                        display_step.presentation.status = self.m.step.WARNING

            if raise_on_failure:
                # Construct failure message. Include both types of failures,
                # regardless of whether we raise purple or red.
                failure_message = []
                if infra_failed_tasks:
                    failure_message.append(
                        "infra failures: {infra_failed_tasks}".format(
                            infra_failed_tasks=", ".join(infra_failed_tasks)
                        )
                    )
                if failed_tasks:
                    failure_message.append(
                        "step failures: {failed_tasks}".format(
                            failed_tasks=", ".join(failed_tasks)
                        )
                    )
                # If there were any infra failures, raise purple.
                if infra_failed_tasks:
                    presentation.status = self.m.step.EXCEPTION
                    exception_type = self.m.step.InfraFailure
                # Otherwise if there were any step failures, raise red.
                elif failed_tasks:
                    presentation.status = self.m.step.FAILURE
                    exception_type = self.m.step.StepFailure
                else:
                    return
                num_failed = len(failed_tasks) + len(infra_failed_tasks)
                truncation_message = (
                    "(failure summary truncated, see individual tasks for full failure "
                    "details)"
                )
                raise exception_type(
                    self.m.buildbucket_util.summary_message(
                        "%s failed\n\n%s"
                        % (pluralize("task", num_failed), "".join(failure_message)),
                        truncation_message,
                        escape_markdown=False,
                    )
                )