recipe_modules/artifacts/api.py

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

from recipe_engine import recipe_api

# The path to the private key used to sign release builds. Only set on release
# builders.
RELEASE_PKEY_PATH = "/etc/release_keys/release_key.pem"

# The retcode that artifactory emits if a transient GCS error occurs.
# Keep in sync with https://fuchsia.googlesource.com/fuchsia/+/HEAD/tools/artifactory/cmd/up.go
TRANSIENT_ERROR_RETCODE = 3


class ArtifactsApi(recipe_api.RecipeApi):
    """API for interacting with build and test artifacts."""

    def __init__(self, *args, **kwargs):
        super(ArtifactsApi, self).__init__(*args, **kwargs)
        # A GCS bucket (str) to which a package repository may be uploaded.
        self.gcs_bucket = None
        # A unique identifier (str) giving a namespace in the GCS bucket under
        # which a package repository may be uploaded.
        self.namespace = None

    def _build_path(self):
        assert self.namespace
        return "builds/%s" % self.namespace

    def _base_url(self, host=None):
        assert self.gcs_bucket
        if host:
            return "http://%s/%s" % (host, self.gcs_bucket)
        return "gs://%s" % self.gcs_bucket

    def debug_symbol_url(self):
        """Returns the URL (str) of the artifact bucket's debug symbol path."""
        return "%s/debug" % self._base_url()

    def build_url(self, host=None):
        """Returns the URL (str) of the uploaded build subdirectory.

        Args:
          host (str or None): The hosting address of the build artifacts; if
            unprovided, the default GCS URL will be constructed.
        """
        assert self.namespace
        return "%s/%s" % (self._base_url(host), self._build_path())

    def package_repo_url(self, host=None):
        """Returns the URL (str) of an uploaded package repository.

        This assumes that we have already upload()ed to this bucket.

        Args:
          host (str or None): The hosting address of the package repository; if
           unprovided, the default GCS URL will be constructed.
        """
        return "%s/packages/repository" % self.build_url(host)

    def package_blob_url(self, host=None):
        """Returns the URL (str) of the blobs of an uploaded package repository.

        Args:
          host (str or None): The hosting address of the package repository; if
           unprovided, the default GCS URL will be constructed.
        """
        return "%s/blobs" % self._base_url(host)

    def image_url(self, host=None):
        """Returns the URL (str) of the uploaded images.

        Args:
          host (str or None): The hosting address of the images; if unprovided, the
           default GCS URL will be constructed.
        """
        return "%s/images" % self.build_url(host)

    def cloud_storage_url(self):
        return self.build_url(host="console.cloud.google.com/storage/browser")

    def upload(
        self,
        step_name,
        build_results,
        upload_host_tests=False,
        sign_artifacts=False,
        timeout_secs=45 * 60,
    ):
        """Uploads built and assembled artifacts.

        This includes package metadata, blobs, keys, and debug binaries.

        Args:
          step_name (str): The name of the step.
          build_results (api.build.FuchsiaBuildResults): The results of a build.
          upload_host_tests (bool): Whether or not to upload host tests.
          sign_artifacts (bool): Whether to sign the artifacts and attach the
            signatures to the uploaded files.
          timeout_secs (int): A timeout for the step in seconds.
        """
        assert self.gcs_bucket and self.namespace
        cmd = [
            build_results.tool("artifactory"),
            "up",
            "-bucket",
            self.gcs_bucket,
            "-namespace",
            self.namespace,
        ]
        if upload_host_tests:
            cmd.append("-upload-host-tests")
        if sign_artifacts:
            cmd.extend(["-pkey", RELEASE_PKEY_PATH])

        cmd.append(build_results.build_dir)
        try:
            step = self.m.step(step_name, cmd, timeout=timeout_secs)
        except self.m.step.StepFailure as e:
            if e.exc_result.retcode == TRANSIENT_ERROR_RETCODE:
                self.m.step.active_result.presentation.status = self.m.step.EXCEPTION
                raise self.m.step.InfraFailure(
                    "Transient GCS error during artifact upload. See '%s' stdout for details."
                    % step_name
                )
            elif e.exc_result.had_timeout:
                self.m.step.active_result.presentation.status = self.m.step.EXCEPTION
                raise self.m.step.InfraFailure(
                    "Artifact upload timed out after %d minutes." % (timeout_secs / 60)
                )
            else:
                raise e

        step.presentation.links["build_artifacts"] = self.cloud_storage_url()

    def download(self, step_name, src, dest, from_root=False):
        """Downloads the specified artifact.

        Args:
          step_name (str): The name of the step.
          src (str): The path of the artifact to download.
          dest (Path): The path to download the artifact to.
          from_root (bool): If true, the `src` will be taken to be relative to
            the GCS bucket (for shared artifacts like blobs and debug binaries),
            else it will be relative to the builds/UUID subdirectory.
        """
        self.m.gsutil.download(
            self.gcs_bucket,
            src if from_root else "%s/%s" % (self._build_path(), src),
            dest,
            name=step_name,
        )