recipe_modules/debug_symbols/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

DEBUGSYMS_VERSION = "git_revision:8ac69fadcd5ab1c38a4b875d79b153825f0b1048"


class DebugSymbolsApi(recipe_api.RecipeApi):
    """APIs for fetching and uploading debug symbols."""

    def fetch_and_upload(self, packages, version, gcs_paths):
        """Fetch CIPD debug symbol packages at a common version, and upload debug
        symbols to GCS.

        Args:
          packages (seq(str)): CIPD packages containing debug symbols to upload.
          version (str): CIPD version common to packages.
          gcs_paths (seq(tuple)): GCS paths as (bucket, namespace) pairs to upload
            debug symbols to.
        """
        ensure_file = self.m.cipd.EnsureFile()
        base_dir = self.m.path.mkdtemp()
        build_id_dirs = []
        for package in packages:
            # Give each package a namespace to avoid tarball filename collisions.
            ensure_file.add_package(package, version, subdir=package)
            build_id_dirs.append(base_dir.join(package))
        self.m.cipd.ensure(base_dir, ensure_file)

        # Discover and unpack archives, if any.
        archives = self.m.file.glob_paths(
            name="find debug symbol archives",
            source=base_dir,
            pattern="**/*.tar.bz2",
            test_data=[base_dir.join("foo", "symbols.tar.bz2")],
        )
        for archive in archives:
            # Extract API requires a unique, non-existent directory.
            output_dir = self.m.path.mkdtemp().join("out")
            self.m.archive.extract(
                step_name="extract %s" % archive,
                archive_file=archive,
                output=output_dir,
            )
            build_id_dirs.append(output_dir)

        for bucket, namespace in gcs_paths:
            self.upload(
                step_name="upload debug symbols to %s" % bucket,
                bucket=bucket,
                namespace=namespace,
                build_id_dirs=build_id_dirs,
            )

    def upload(self, step_name, bucket, build_id_dirs, namespace=None):
        """Walk a list of .build-id dirs and upload found debug symbols to GCS.

        Args:
          step_name (str): The name of the step produced.
          bucket (str): GCS bucket to upload symbols to.
          build_id_dirs (seq(Path)): Paths to .build-id dirs.
          namespace (str): GCS namespace to upload symbols to.
        """
        args = [self._debugsyms_tool, "upload", "-bucket", bucket]
        if namespace:
            args.extend(["-namespace", namespace])
        args.extend(build_id_dirs)
        return self.m.step(step_name, args)

    @property
    def _debugsyms_tool(self):
        return self.m.cipd.ensure_tool(
            "fuchsia/tools/debugsyms/${platform}", DEBUGSYMS_VERSION
        )