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().__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_from_manifest(
        self,
        step_name,
        upload_manifest_json_input,
        sign_artifacts=False,
        timeout_secs=45 * 60,
    ):
        """Upload build artifacts to GCS per the input upload manifest.

        Args:
            step_name (str): Name of the step.
            upload_manifest_json_input (Path): Path of upload manifest as JSON.
            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 = [
            self._gcs_util_tool,
            "up",
            "-bucket",
            self.gcs_bucket,
            "-namespace",
            self.build_path(),
            "-manifest-path",
            upload_manifest_json_input,
        ]
        if sign_artifacts:
            cmd.extend(["-pkey", RELEASE_PKEY_PATH])

        try:
            step = self.m.step(step_name, cmd, timeout=timeout_secs)
        except self.m.step.StepFailure as e:
            if e.exc_result.was_cancelled:
                raise
            if 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)
                )
            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
                )
            raise

        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,
        )

    @property
    def _gcs_util_tool(self):
        return self.m.ensure_tool("gcs-util", self.resource("tool_manifest.json"))