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:cd8fdd61ae1e1553987cdaffdf4ac1b9396dab66"


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

    def fetch_and_upload(self, packages, version, buckets):
        """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.
          buckets (seq(str)): GCS buckets to upload
            debug symbols to.

        Returns:
          list(Path): Paths to .build_id directories.
        """
        ensure_file = self.m.cipd.EnsureFile()
        base_dir = self.m.path.mkdtemp("debug_symbols")
        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("extract").join("out")
            self.m.archive.extract(
                step_name="extract %s" % archive,
                archive_file=archive,
                output=output_dir,
            )
            build_id_dirs.append(output_dir)

        self.upload(
            step_name="upload debug symbols",
            build_id_dirs=build_id_dirs,
            buckets=buckets,
        )
        return build_id_dirs

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

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

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