lib/xbuddy/xbuddy.py

# Copyright 2013 The ChromiumOS Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""Main module for parsing and interpreting XBuddy paths for the devserver."""

import configparser
import datetime
import distutils.version  # pylint: disable=import-error,no-name-in-module
import logging
import operator
import os
import re
import shutil
import sys
import threading
import time
from typing import List, NamedTuple, Optional, Tuple
import urllib

from chromite.lib import constants
from chromite.lib import gs
from chromite.lib import image_lib
from chromite.lib import osutils
from chromite.lib import path_util
from chromite.lib.xbuddy import artifact_info
from chromite.lib.xbuddy import build_artifact
from chromite.lib.xbuddy import common_util
from chromite.lib.xbuddy import devserver_constants
from chromite.lib.xbuddy import downloader


# xBuddy config constants
CONFIG_FILE = "xbuddy_config.ini"
SHADOW_CONFIG_FILE = "shadow_xbuddy_config.ini"
PATH_REWRITES = "PATH_REWRITES"
GENERAL = "GENERAL"
LOCATION_SUFFIXES = "LOCATION_SUFFIXES"

# Path for shadow config in chroot.
CHROOT_SHADOW_DIR = "/mnt/host/source/src/platform/dev"

# XBuddy aliases
TEST = "test"
BASE = "base"
DEV = "dev"
FULL = "full_payload"
DELTA = "delta_payload"
SIGNED = "signed"
RECOVERY = "recovery"
STATEFUL = "stateful"
AUTOTEST = "autotest"
FACTORY_SHIM = "factory_shim"

# Local build constants
ANY = "ANY"
LATEST = "latest"
LOCAL = "local"
REMOTE = "remote"

# TODO(sosa): Fix a lot of assumptions about these aliases. There is too much
# implicit logic here that's unnecessary. What should be done:
# 1) Collapse Alias logic to one set of aliases for xbuddy (not local/remote).
# 2) Do not use zip when creating these dicts. Better to not rely on ordering.
# 3) Move alias/artifact mapping to a central module rather than having it here.
# 4) Be explicit when things are missing i.e. no dev images in image.zip.

LOCAL_ALIASES = [
    TEST,
    DEV,
    BASE,
    RECOVERY,
    FACTORY_SHIM,
    FULL,
    STATEFUL,
    ANY,
]

LOCAL_FILE_NAMES = [
    devserver_constants.TEST_IMAGE_FILE,
    devserver_constants.IMAGE_FILE,
    devserver_constants.BASE_IMAGE_FILE,
    devserver_constants.RECOVERY_IMAGE_FILE,
    devserver_constants.FACTORY_SHIM_IMAGE_FILE,
    devserver_constants.UPDATE_FILE,
    devserver_constants.STATEFUL_FILE,
    None,  # For ANY.
]

LOCAL_ALIAS_TO_FILENAME = dict(zip(LOCAL_ALIASES, LOCAL_FILE_NAMES))

# Google Storage constants
GS_ALIASES = [
    TEST,
    BASE,
    RECOVERY,
    SIGNED,
    FACTORY_SHIM,
    FULL,
    DELTA,
    STATEFUL,
    AUTOTEST,
]

ARTIFACTS = [
    artifact_info.TEST_IMAGE,
    artifact_info.BASE_IMAGE,
    artifact_info.RECOVERY_IMAGE,
    artifact_info.SIGNED_IMAGE,
    artifact_info.FACTORY_SHIM_IMAGE,
    artifact_info.FULL_PAYLOAD,
    artifact_info.DELTA_PAYLOAD,
    artifact_info.STATEFUL_PAYLOAD,
    artifact_info.AUTOTEST,
]

GS_ALIAS_TO_ARTIFACT = dict(zip(GS_ALIASES, ARTIFACTS))

LATEST_OFFICIAL = "latest-official"

RELEASE = "-release"

DEFAULT_STATIC_DIR = os.path.join(constants.SOURCE_ROOT, "devserver", "static")

_TIMESTAMP_DELIMITER = "SLASH"
_XBUDDY_TIMESTAMP_DIR = "xbuddy_UpdateTimestamps"


class XBuddyComponents(NamedTuple):
    """Container for XBuddy URI components."""

    image_type: str
    board: Optional[str]
    version: str
    is_local: bool


class XBuddyException(Exception):
    """Exception classes used by this module."""


class XBuddyInvalidSchemeException(XBuddyException):
    """Exception raised for invalid URI scheme in XBuddy URI."""


def update_timestamp(timestamp_dir: os.PathLike, build_id: os.PathLike) -> None:
    """Update timestamp file of build with build_id."""
    common_util.MkDirP(timestamp_dir)
    logging.debug("Updating timestamp for %s", build_id)
    osutils.Touch(os.path.join(timestamp_dir, build_to_timestamp(build_id)))


def timestamp_to_build(timestamp_filename: os.PathLike) -> os.PathLike:
    """Perform delimiter conversion on a timestamp file to get an image path."""
    return timestamp_filename.replace(_TIMESTAMP_DELIMITER, "/")


def build_to_timestamp(build_path: os.PathLike) -> os.PathLike:
    """Perform delimiter conversion on an image path to get a timestamp file."""
    return build_path.replace("/", _TIMESTAMP_DELIMITER)


def find_any(local_dir: os.PathLike) -> Optional[str]:
    """Returns the image_type for ANY given the local_dir."""
    test_image = os.path.join(local_dir, devserver_constants.TEST_IMAGE_FILE)
    dev_image = os.path.join(local_dir, devserver_constants.IMAGE_FILE)
    base_image = os.path.join(local_dir, devserver_constants.BASE_IMAGE_FILE)
    # Prioritize test images over dev images.
    if os.path.exists(test_image):
        return "test"

    if os.path.exists(dev_image):
        return "dev"

    if os.path.exists(base_image):
        logging.warning(
            "Using base image as test and dev images were not found. "
            "This will likely cause cros flash to fail leading to "
            "needing to repair the device with a USB stick."
        )
        return "base"

    return None


class XBuddy:
    """Class that manages image retrieval and caching by the devserver.

    Image retrieval by xBuddy path:
      XBuddy accesses images and artifacts that it stores using an xBuddy
      path of the form: board/version/alias
      The primary xbuddy.Get call retrieves the correct artifact or url to where
      the artifacts can be found.

    Image caching:
      Images and other artifacts are stored identically to how they would have
      been if devserver's stage rpc was called and the xBuddy cache replaces
      build versions on a LRU basis. Timestamps are maintained by last accessed
      times of representative files in the a directory in the static serve
      directory (XBUDDY_TIMESTAMP_DIR).

    Private class members:
      _true_values: used for interpreting boolean values
      _staging_thread_count: track download requests
      _timestamp_folder: directory with empty files standing in as timestamps
                          for each image currently cached by xBuddy
    """

    _true_values = ["true", "t", "yes", "y"]

    # Number of threads that are staging images.
    _staging_thread_count = 0
    # Lock used to lock increasing/decreasing count.
    _staging_thread_count_lock = threading.Lock()

    def __init__(
        self,
        manage_builds: bool = False,
        board: str = "",
        version: str = "",
        images_dir: os.PathLike = "",
        static_dir: os.PathLike = DEFAULT_STATIC_DIR,
        gsutil_bin: Optional[os.PathLike] = None,
    ):
        """Initialize an XBuddy image file manager.

        Args:
            manage_builds: Indicates whether XBuddy is managing local builds.
            board: The name of the board of interest.
            version: The version to fill in when rewriting paths; usually either
                a specific version number or an alias like LATEST.
            images_dir: The (local) directory where images will be cached.
            static_dir: A utility folder used for caching operations; of
                particular interest is the timestamp folder located in a subdir,
                which is the tracking mechanism for LRU caching and retrieval.
            gsutil_bin: The absolute path to the `gsutil` binary.
        """
        self.config = self._ReadConfig()
        self._manage_builds = manage_builds or self._ManageBuilds()
        self._board = board
        self._version = version
        self.static_dir = static_dir
        self._timestamp_folder = os.path.join(
            self.static_dir, _XBUDDY_TIMESTAMP_DIR
        )
        self.images_dir = images_dir or os.path.join(
            constants.SOURCE_ROOT, "src/build/images"
        )

        cache_user = "chronos" if common_util.IsRunningOnMoblab() else None
        self._ctx = gs.GSContext(cache_user=cache_user, gsutil_bin=gsutil_bin)

        common_util.MkDirP(self._timestamp_folder)

    @classmethod
    def ParseBoolean(cls, boolean_string: Optional[str]) -> bool:
        """Evaluate a string to a boolean value"""
        if boolean_string:
            return boolean_string.lower() in cls._true_values
        else:
            return False

    def _ReadConfig(self) -> configparser.ConfigParser:
        """Read xbuddy config from ini files.

        Reads the base config from xbuddy_config.ini, and then merges in the
        shadow config from shadow_xbuddy_config.ini

        Returns:
            The merged configuration.

        Raises:
            XBuddyException if the config file is missing.
        """
        devserver_dir = os.path.dirname(os.path.abspath(sys.argv[0]))
        config_file = os.path.join(devserver_dir, CONFIG_FILE)
        xbuddy_config = configparser.ConfigParser()
        if os.path.exists(config_file):
            xbuddy_config.read(config_file)
        else:
            # Get the directory of xbuddy.py file.
            file_dir = os.path.dirname(os.path.realpath(__file__))
            # Read the default xbuddy_config.ini from the directory.
            xbuddy_config.read(os.path.join(file_dir, CONFIG_FILE))

        # Read the shadow file if there is one.
        if os.path.isdir(CHROOT_SHADOW_DIR):
            shadow_config_file = os.path.join(
                CHROOT_SHADOW_DIR, SHADOW_CONFIG_FILE
            )
        else:
            shadow_config_file = os.path.join(devserver_dir, SHADOW_CONFIG_FILE)

        logging.debug(
            "Using shadow config file stored at %s", shadow_config_file
        )
        if os.path.exists(shadow_config_file):
            shadow_xbuddy_config = configparser.ConfigParser()
            shadow_xbuddy_config.read(shadow_config_file)

            # Merge shadow config in.
            sections = shadow_xbuddy_config.sections()
            for s in sections:
                if not xbuddy_config.has_section(s):
                    xbuddy_config.add_section(s)
                options = shadow_xbuddy_config.options(s)
                for o in options:
                    val = shadow_xbuddy_config.get(s, o)
                    xbuddy_config.set(s, o, val)

        return xbuddy_config

    def _ManageBuilds(self) -> bool:
        """Check if xBuddy is managing local builds using the current config."""
        try:
            return self.ParseBoolean(self.config.get(GENERAL, "manage_builds"))
        except configparser.Error:
            return False

    def _Capacity(self) -> int:
        """Gets the xbuddy capacity from the current config."""
        try:
            return int(self.config.get(GENERAL, "capacity"))
        except configparser.Error:
            return 5

    def LookupAlias(
        self,
        alias: str,
        board: Optional[str] = None,
        version: Optional[str] = None,
    ) -> Tuple[os.PathLike, str]:
        """Given the full xbuddy config, look up an alias for path rewrite.

        Args:
            alias: The xbuddy path that could be one of the aliases in the
                rewrite table.
            board: The board to fill in with when paths are rewritten. Can be
                from the update request xml or the default board from devserver.
                If None, defers to the value given during XBuddy initialization.
            version: The version to fill in when rewriting paths. Could be a
                specific version number or a version alias like LATEST. If None,
                defers to the value given during XBuddy initialization, or
                LATEST.

        Returns:
            A pair (val, suffix) where val is the rewritten path, or the
            original string if no rewrite was found; and suffix is the assigned
            location suffix, or the default suffix if none was found.
        """
        try:
            suffix = self.config.get(LOCATION_SUFFIXES, alias)
        except configparser.Error:
            suffix = RELEASE

        try:
            val = self.config.get(PATH_REWRITES, alias)
        except configparser.Error:
            # No alias lookup found. Return original path.
            val = None

        if not (val and val.strip()):
            val = alias
        else:
            # The found value is not an empty string.
            # Fill in the board and version.
            val = val.replace("BOARD", "%(board)s")
            val = val.replace("VERSION", "%(version)s")
            val = val % {
                "board": board or self._board,
                "version": version or self._version or LATEST,
            }

        return val, suffix

    def _LookupOfficial(
        self, board: str, suffix: str, image_dir: Optional[os.PathLike] = None
    ) -> os.PathLike:
        """Check LATEST-master for the version number of interest."""
        logging.debug("Checking gs for latest %s-%s image", board, suffix)
        image_dir = (image_dir or devserver_constants.GS_IMAGE_DIR).rstrip("/")

        version = None
        for f in (
            devserver_constants.GS_LATEST_MAIN,
            devserver_constants.GS_LATEST_MASTER,
        ):
            latest_addr = f % {
                "image_dir": image_dir,
                "board": board,
                "suffix": suffix,
            }

            try:
                # Full release + version is in the LATEST file.
                version = self._ctx.Cat(latest_addr, encoding="utf-8")
                return devserver_constants.IMAGE_DIR % {
                    "board": board,
                    "suffix": suffix,
                    "version": version,
                }
            except gs.GSContextException as e:
                logging.warning(
                    "Failed to look up file %s with error %s. ignoring", f, e
                )

        raise build_artifact.ArtifactDownloadError(
            "Cannot look up the official image version."
        )

    def _LS(
        self, path: os.PathLike, list_subdirectory: bool = False
    ) -> List[os.PathLike]:
        """Does a directory listing of the given gs path.

        Args:
            path: directory location on google storage to check.
            list_subdirectory: whether to only list subdirectory for |path|.

        Returns:
            A list of paths that matched |path|.
        """
        if list_subdirectory:
            return self._ctx.DoCommand(
                ["ls", "-d", "--", path], stdout=True
            ).stdout.splitlines()
        else:
            return self._ctx.LS(path)

    def _GetLatestVersionFromGsDir(
        self,
        path: os.PathLike,
        list_subdirectory: bool = False,
        with_release: bool = True,
    ) -> str:
        """Get most recent version number found in a google storage directory.

        This lists out the contents of the given GS bucket or regex to GS
        buckets, and tries to grab the newest version found in the directory
        names.

        Args:
            path: directory location on google storage to check.
            list_subdirectory: whether to only list subdirectory for |path|.
            with_release: whether versions include a release milestone (e.g.
                R12).

        Returns:
            The most recent version number found.
        """
        list_result = self._LS(path, list_subdirectory=list_subdirectory)
        dir_names = [os.path.basename(p.rstrip("/")) for p in list_result]
        try:
            versions_re = re.compile(
                devserver_constants.VERSION_RE
                if with_release
                else devserver_constants.VERSION
            )
            versions = [d for d in dir_names if versions_re.match(d)]
            latest_version = max(versions, key=distutils.version.LooseVersion)
        except ValueError:
            raise gs.GSContextException(
                "Failed to find most recent builds at %s" % path
            )

        return latest_version

    def _LookupChannel(
        self,
        board: str,
        suffix: str,
        channel: str = "stable",
        image_dir: Optional[os.PathLike] = None,
    ) -> str:
        """Check the channel folder for the version number of interest."""
        # Get all names in channel dir. Get 10 highest directories by version.
        logging.debug(
            "Checking channel '%s' for latest '%s' image", channel, board
        )
        # Due to historical reasons, gs://chromeos-releases uses
        # daisy-spring as opposed to the board name daisy_spring. Convert
        # he board name for the lookup.
        channel_dir = devserver_constants.GS_CHANNEL_DIR % {
            "channel": channel,
            "board": re.sub("_", "-", board),
        }
        latest_version = self._GetLatestVersionFromGsDir(
            channel_dir, with_release=False
        )

        # Figure out release number from the version number.
        image_url = devserver_constants.IMAGE_DIR % {
            "board": board,
            "suffix": suffix,
            "version": "R*" + latest_version,
        }
        image_dir = (image_dir or devserver_constants.GS_IMAGE_DIR).rstrip("/")
        gs_url = os.path.join(image_dir, image_url)

        # There should only be one match on cros-image-archive.
        full_version = self._GetLatestVersionFromGsDir(
            gs_url, list_subdirectory=True
        )

        return devserver_constants.IMAGE_DIR % {
            "board": board,
            "suffix": suffix,
            "version": full_version,
        }

    def _LookupVersion(self, board: str, suffix: str, version: str) -> str:
        """Search GS image releases for the highest match to a version."""
        # Build the pattern for GS to match.
        logging.debug(
            "Checking gs for latest '%s' image with prefix '%s'", board, version
        )
        image_url = devserver_constants.IMAGE_DIR % {
            "board": board,
            "suffix": suffix,
            "version": version + "*",
        }
        image_dir = os.path.join(devserver_constants.GS_IMAGE_DIR, image_url)

        # Grab the newest version of the ones matched.
        full_version = self._GetLatestVersionFromGsDir(
            image_dir, list_subdirectory=True
        )
        return devserver_constants.IMAGE_DIR % {
            "board": board,
            "suffix": suffix,
            "version": full_version,
        }

    def _RemoteBuildId(
        self, board: str, suffix: str, version: str
    ) -> os.PathLike:
        """Returns the remote build_id for the given board and version.

        Raises:
            XBuddyException: If we failed to resolve the version to a valid
            build_id.
        """
        build_id_as_is = devserver_constants.IMAGE_DIR % {
            "board": board,
            "suffix": "",
            "version": version,
        }
        build_id_suffix = devserver_constants.IMAGE_DIR % {
            "board": board,
            "suffix": suffix,
            "version": version,
        }
        # Return the first path that exists. We assume that what the user typed
        # is better than with a default suffix added i.e. x86-generic/blah is
        # more valuable than x86-generic-release/blah.
        for build_id in build_id_as_is, build_id_suffix:
            try:
                version = self._ctx.LS(
                    "%s/%s" % (devserver_constants.GS_IMAGE_DIR, build_id)
                )
                return build_id
            except gs.GSContextException as e:
                if common_util.IsAnonymousCaller(e):
                    logging.error(
                        "Anonymous caller cannot list chromeos image archive"
                    )
                    return build_id_as_is
                continue

        raise XBuddyException(
            "Could not find remote build_id for %s %s" % (board, version)
        )

    def _ResolveBuildVersion(
        self, board: str, suffix: str, base_version: str
    ) -> str:
        """Check LATEST-<base_version> and returns a full build version."""
        logging.debug(
            "Checking gs for full version for %s of %s", base_version, board
        )
        # TODO(garnold) We might want to accommodate version prefixes and pick
        #   the most recent found, as done in _LookupVersion().
        latest_addr = devserver_constants.GS_LATEST_BASE_VERSION % {
            "image_dir": devserver_constants.GS_IMAGE_DIR,
            "board": board,
            "suffix": suffix,
            "base_version": base_version,
        }
        # Full release + version is in the LATEST file.
        return self._ctx.Cat(latest_addr)

    def _ResolveVersionToBuildIdAndChannel(
        self,
        board: str,
        suffix: str,
        version: str,
        image_dir: Optional[os.PathLike] = None,
    ) -> Tuple[os.PathLike, str]:
        """Handle version aliases for remote payloads in GS.

        Args:
            board: as specified in the original call. (i.e. x86-generic, parrot)
            suffix: The location suffix, to be added to board name.
            version: as entered in the original call. can be
                {TBD, 0. some custom alias as defined in a config file}
                1. fully qualified build version.
                2. latest
                3. latest-{channel}
                4. latest-official-{board suffix}
                5. version prefix (i.e. RX-Y.X, RX-Y, RX)
            image_dir: image directory to check in Google Storage. If none,
                the default bucket is used.

        Returns:
            Tuple of (Location where the image dir is actually found on GS
            (build_id), best guess for the channel).

        Raises:
            XBuddyException: If we failed to resolve the version to a valid url.
        """
        # Only the last segment of the alias is variable relative to the rest.
        version_tuple = version.rsplit("-", 1)

        if re.match(devserver_constants.VERSION_RE, version):
            return self._RemoteBuildId(board, suffix, version), None
        elif re.match(devserver_constants.VERSION, version):
            raise XBuddyException(
                '"%s" is not valid. Should provide the fully '
                'qualified version with a version prefix "RX-" '
                "due to crbug.com/585914" % version
            )
        elif version == LATEST_OFFICIAL:
            # latest-official --> LATEST build in board-release
            return (
                self._LookupOfficial(board, suffix, image_dir=image_dir),
                None,
            )
        elif version_tuple[0] == LATEST_OFFICIAL:
            # latest-official-{suffix} --> LATEST build in board-{suffix}
            return (
                self._LookupOfficial(
                    board, version_tuple[1], image_dir=image_dir
                ),
                None,
            )
        elif version == LATEST:
            # latest --> latest build on stable channel
            return (
                self._LookupChannel(board, suffix, image_dir=image_dir),
                "stable",
            )
        elif version_tuple[0] == LATEST:
            if re.match(devserver_constants.VERSION_RE, version_tuple[1]):
                # latest-R* --> most recent qualifying build
                return (
                    self._LookupVersion(board, suffix, version_tuple[1]),
                    None,
                )
            else:
                # latest-{channel} --> latest build within that channel
                return (
                    self._LookupChannel(
                        board,
                        suffix,
                        channel=version_tuple[1],
                        image_dir=image_dir,
                    ),
                    version_tuple[1],
                )
        else:
            # The given version doesn't match any known patterns.
            raise XBuddyException(
                "Version %s unknown. Can't find on GS." % version
            )

    def _GetLatestLocalVersion(self, board: str) -> str:
        """Get the version of the latest image built by `cros build-image`.

        Updates the symlink reference within the xBuddy static dir to point to
        the real image dir in the local /build/images directory.

        Args:
            board: board that image was built for.

        Returns:
            The discovered version of the image.

        Raises:
            XBuddyException: if neither test nor dev image was found in latest
            built directory.
        """
        latest_local_dir = image_lib.GetLatestImageLink(board)
        if not latest_local_dir or not os.path.exists(latest_local_dir):
            raise XBuddyException(
                f"No builds found for {board}. Did you run "
                f"`cros build-image --board {board}`?"
            )

        # Assume that the version number is the name of the directory.
        return os.path.basename(os.path.realpath(latest_local_dir))

    def _SyncRegistryWithBuildImages(self) -> None:
        """Crawl images_dir for build_ids of images from `cros build-image`.

        This will find images and symlink them in xBuddy's static dir so that
        xBuddy's cache can serve them.
        If xBuddy's _manage_builds option is on, then a timestamp will also be
        generated, and xBuddy will clear them from the directory they are in, as
        necessary.
        """
        if not os.path.isdir(self.images_dir):
            # Skip syncing if images_dir does not exist.
            logging.error(
                "Cannot find %s; skip syncing image registry.", self.images_dir
            )
            return

        build_ids = []
        for b in os.listdir(self.images_dir):
            # Ignore random files in the build dir.
            board_dir = os.path.join(self.images_dir, b)
            if not os.path.isdir(board_dir):
                continue

            # Ensure we have directories to track all boards in build/images
            common_util.MkDirP(os.path.join(self.static_dir, b))
            build_ids.extend(
                [
                    "/".join([b, v])
                    for v in os.listdir(board_dir)
                    if not v == LATEST
                ]
            )

        # Symlink undiscovered images, and update timestamps if manage_builds is
        # on.
        for build_id in build_ids:
            link = os.path.join(self.static_dir, build_id)
            target = os.path.join(self.images_dir, build_id)
            osutils.SafeSymlink(target, link)
            if self._manage_builds:
                update_timestamp(self._timestamp_folder, build_id)

    def _ListBuildTimes(self) -> List[Tuple[os.PathLike, datetime.timedelta]]:
        """Returns the currently cached builds and their last access timestamp.

        Returns:
            list of tuples that matches xBuddy build/version to timestamps in
            long
        """
        # Update currently cached builds.
        build_dict = {}

        for f in os.listdir(self._timestamp_folder):
            last_accessed = os.path.getmtime(
                os.path.join(self._timestamp_folder, f)
            )
            build_id = timestamp_to_build(f)
            stale_time = datetime.timedelta(
                seconds=(time.time() - last_accessed)
            )
            build_dict[build_id] = stale_time
        return_tup = sorted(build_dict.items(), key=operator.itemgetter(1))
        return return_tup

    def _Download(
        self, gs_url: str, artifacts: List[str], build_id: os.PathLike
    ) -> List[List[str]]:
        """Download the artifacts from the given gs_url.

        Returns:
            A list containing lists of downloaded files for each artifact.
            e.g.: artifacts = ['a', 'b'] -> return [['f1', 'f2'], ['f3']]

        Raises:
            build_artifact.ArtifactDownloadError: If we failed to download the
            artifact.
        """
        with XBuddy._staging_thread_count_lock:
            XBuddy._staging_thread_count += 1
        try:
            logging.debug("Downloading %s from %s", artifacts, gs_url)
            dl = downloader.GoogleStorageDownloader(
                self.static_dir, gs_url, build_id
            )
            factory = build_artifact.ChromeOSArtifactFactory(
                dl.GetBuildDir(), artifacts, [], dl.GetBuild()
            )
            dl.Download(factory)
            downloaded_artifacts = factory.RequiredArtifacts()
            return [x.StagedFiles() for x in downloaded_artifacts]
        finally:
            with XBuddy._staging_thread_count_lock:
                XBuddy._staging_thread_count -= 1

    def CleanCache(self) -> None:
        """Delete all builds besides the newest N builds"""
        if not self._manage_builds:
            return
        cached_builds = [e[0] for e in self._ListBuildTimes()]
        logging.debug("In cache now: %s", cached_builds)

        for b in range(self._Capacity(), len(cached_builds)):
            b_path = cached_builds[b]
            logging.debug("Clearing '%s' from cache", b_path)

            time_file = os.path.join(
                self._timestamp_folder, build_to_timestamp(b_path)
            )
            os.unlink(time_file)
            clear_dir = os.path.join(self.static_dir, b_path)
            try:
                # Handle symlinks, in the case of links to local builds if
                # enabled.
                if os.path.islink(clear_dir):
                    target = os.readlink(clear_dir)
                    logging.debug("Deleting locally built image at %s", target)

                    os.unlink(clear_dir)
                    if os.path.exists(target):
                        shutil.rmtree(target)
                elif os.path.exists(clear_dir):
                    logging.debug("Deleting downloaded image at %s", clear_dir)
                    shutil.rmtree(clear_dir)

            except OSError as e:
                raise XBuddyException(f"Failed to clear {clear_dir}: {e}")

    def _TranslateSignedGSUrl(
        self, build_id: os.PathLike, channel: str = None
    ) -> os.PathLike:
        """Translate the GS URL to be able to find signed images.

        Args:
            build_id: Path to the image or update directory on the devserver or
                in Google Storage. e.g. 'x86-generic/R26-4000.0.0'
            channel: The channel for the image. If none, it tries to guess it in
                order of stability.

        Returns:
            The GS URL for the directory where the signed image can be found.

        Raises:
            build_artifact.ArtifactDownloadError: If we failed to download the
            artifact.
        """
        match = re.match(r"^([^/]+?)(?:-release)?/R\d+-(.*)$", build_id)

        channels = []
        if channel:
            channels.append(channel)
        else:
            # Attempt to enumerate all channels, in order of stability.
            channels.extend(devserver_constants.CHANNELS[::-1])

        for c in channels:
            image_dir = devserver_constants.GS_CHANNEL_DIR % {
                "channel": c,
                "board": match.group(1),
            }
            gs_url = os.path.join(image_dir, match.group(2))
            try:
                self._LS(gs_url)
                return gs_url
            except gs.GSNoSuchKey:
                continue
        raise build_artifact.ArtifactDownloadError(
            "Could not find signed image URL for %s in Google Storage"
            % build_id
        )

    def _GetFromGS(
        self,
        build_id: os.PathLike,
        image_type: str,
        image_dir: Optional[os.PathLike] = None,
        channel: Optional[str] = None,
    ) -> List[os.PathLike]:
        """Check if the artifact is available locally. Download from GS if not.

        Args:
            build_id: Path to the image or update directory on the devserver or
                in Google Storage. e.g. 'x86-generic/R26-4000.0.0'
            image_type: Image type to download. Look at aliases at top of file
                for options.
            image_dir: Google Storage image archive to search in if requesting a
                remote artifact. If none uses the default bucket.
            channel: The channel for the image. If none, it tries to guess it in
                order of stability.

        Returns:
            The list of files downloaded for the given image_type.

        Raises:
            build_artifact.ArtifactDownloadError: If we failed to download the
            artifact.
        """
        artifact = GS_ALIAS_TO_ARTIFACT[image_type]
        if image_type == SIGNED:
            gs_url = self._TranslateSignedGSUrl(build_id, channel=channel)
        else:
            image_dir = (image_dir or devserver_constants.GS_IMAGE_DIR).rstrip(
                "/"
            )
            gs_url = os.path.join(image_dir, build_id)

        return self._Download(gs_url, [artifact], build_id)[0]

    def _GetArtifact(
        self,
        path_list: List[str],
        board: Optional[str] = None,
        version: Optional[str] = None,
        lookup_only: bool = False,
        image_dir: Optional[os.PathLike] = None,
    ) -> Tuple[os.PathLike, os.PathLike]:
        """Interpret an xBuddy path and return directory/file_name to resource.

        Note board can be passed that in but by default if self._board is set,
        that is used rather than board.

        Args:
            path_list: [board, version, alias] as split from the xbuddy call
                url.
            board: Board whose artifacts we are looking for. Only used if no
                board was given during XBuddy initialization.
            version: Version whose artifacts we are looking for. Used if no
                version was given during XBuddy initialization. If None, defers
                to LATEST.
            lookup_only: If true just look up the artifact, if False stage it.
                TODO(ahassani): If true, it should also return the file name.
            image_dir: Google Storage image archive to search in if requesting a
                remote artifact. If none uses the default bucket.

        Returns:
            build_id: Path to the image or update directory on the devserver or
                in Google Storage. e.g. 'x86-generic/R26-4000.0.0'
            file_name: of the artifact in the build_id directory.

        Raises:
            XBuddyException: if the path could not be translated.
            build_artifact.ArtifactDownloadError: if we failed to download the
                artifact.
        """
        path = "/".join(path_list)
        default_board = self._board or board
        default_version = self._version or version or LATEST
        # Rewrite the path if there is an appropriate default.
        path, suffix = self.LookupAlias(
            path, board=default_board, version=default_version
        )
        # Parse the path.
        xbuddy_components = InterpretPath(path, default_board, default_version)
        file_name = None
        if xbuddy_components.is_local:
            # Get a local image.
            version = (
                self._GetLatestLocalVersion(xbuddy_components.board)
                if xbuddy_components.version == LATEST
                else xbuddy_components.version
            )

            build_id = os.path.join(xbuddy_components.board, version)
            artifact_dir = os.path.join(self.static_dir, build_id)
            image_type = (
                find_any(artifact_dir)
                if xbuddy_components.image_type == ANY
                else xbuddy_components.image_type
            )

            # If there was an image type discovered, get the file name
            # otherwise, just return with no file name.
            if image_type:
                file_name = os.path.join(
                    artifact_dir, LOCAL_ALIAS_TO_FILENAME[image_type]
                )
                if not os.path.exists(file_name):
                    raise XBuddyException(
                        "Local %s artifact not in static_dir at %s"
                        % (image_type, file_name)
                    )
        else:
            # Get a remote image.
            if xbuddy_components.image_type not in GS_ALIASES:
                raise XBuddyException(
                    "Bad remote image type: %s. Use one of: %s"
                    % (xbuddy_components.image_type, GS_ALIASES)
                )
            build_id, channel = self._ResolveVersionToBuildIdAndChannel(
                xbuddy_components.board,
                suffix,
                xbuddy_components.version,
                image_dir=image_dir,
            )
            logging.debug("Resolved version %s to %s.", version, build_id)

            if not lookup_only:
                file_name = self._GetFromGS(
                    build_id,
                    xbuddy_components.image_type,
                    image_dir=image_dir,
                    channel=channel,
                )[0]

        return build_id, file_name

    ############################ BEGIN PUBLIC METHODS

    def Translate(
        self,
        path_list: List[str],
        board: Optional[str] = None,
        version: Optional[str] = None,
        image_dir: Optional[os.PathLike] = None,
    ) -> Tuple[os.PathLike, os.PathLike]:
        """Translates an xBuddy path to a real path to artifact if it exists.

        Equivalent to the Get call, minus downloading and updating timestamps,

        Args:
            path_list: [board, version, alias] as split from the xbuddy call
                url.
            board: Board whose artifacts we are looking for. If None, use the
                board XBuddy was initialized to use.
            version: Version whose artifacts we are looking for. If None, use
                the version XBuddy was initialized with, or LATEST.
            image_dir: image directory to check in Google Storage. If none, the
                default bucket is used.

        Returns:
            build_id: Path to the image or update directory on the devserver.
                e.g. 'x86-generic/R26-4000.0.0'
                The returned path is always the path to the directory within
                static_dir, so it is always the build_id of the image.
            file_name: The file name of the artifact. Can take any of the file
                values in devserver_constants.
                e.g. 'chromiumos_test_image.bin' or 'update.gz' if the path list
                specified 'test' or 'full_payload' artifacts, respectively.

        Raises:
            XBuddyException: if the path couldn't be translated
        """
        self._SyncRegistryWithBuildImages()
        return self._GetArtifact(
            path_list,
            board=board,
            version=version,
            lookup_only=True,
            image_dir=image_dir,
        )

    def Get(
        self, path_list: List[str], image_dir: Optional[os.PathLike] = None
    ) -> Tuple[os.PathLike, os.PathLike]:
        """The full xBuddy call, returns resource specified by path_list.

        Please see devserver.py:xbuddy for full documentation.

        Args:
            path_list: [board, version, alias] as split from the xbuddy call
                url.
            image_dir: image directory to check in Google Storage. If none, the
                default bucket is used.

        Returns:
            build_id: Path to the image or update directory on the devserver.
                e.g. 'x86-generic/R26-4000.0.0'
                The returned path is always the path to the directory within
                static_dir, so it is always the build_id of the image.
            file_name: The file name of the artifact. Can take any of the file
                values in devserver_constants.
                e.g. 'chromiumos_test_image.bin' or 'update.gz' if the path list
                specified 'test' or 'full_payload' artifacts, respectively.

        Raises:
            XBuddyException: if the path could not be translated
            build_artifact.ArtifactDownloadError: if we failed to download the
                artifact.
        """
        self._SyncRegistryWithBuildImages()
        build_id, file_name = self._GetArtifact(path_list, image_dir=image_dir)
        update_timestamp(self._timestamp_folder, build_id)
        # TODO(joyc): Run in separate thread.
        self.CleanCache()

        logging.debug(
            "Returning build id: %s and path to payload: %s",
            build_id,
            file_name,
        )
        return build_id, file_name


def InterpretPath(
    path: str, default_board: str = None, default_version: str = None
) -> XBuddyComponents:
    """Split and return the pieces of an xBuddy path name

    Args:
        path: the path xBuddy Get was called with.
        default_board: board to use in case board isn't in path.
        default_version: Version to use in case version isn't in path.

    Returns:
        tuple of (image_type, board, version, whether the path is local)

    Raises:
        XBuddyException: if the path can't be resolved into valid components
    """
    if path.lower().startswith("xbuddy://"):
        path = path[9:]
    path_list = [p for p in path.split("/") if p]

    # Do the stuff that is well known first. We know that if paths have a
    # image_type, it must be one of the GS/LOCAL aliases and it must be at the
    # end. Similarly, local/remote are well-known and must start the path list.
    # Default to remote for chrome checkout.
    is_local = (
        path_util.DetermineCheckout().type != path_util.CHECKOUT_TYPE_GCLIENT
    )
    if path_list and path_list[0] in (REMOTE, LOCAL):
        is_local = path_list.pop(0) == LOCAL

    # Default image type is determined by remote vs. local.
    image_type = ANY if is_local else TEST

    if path_list and path_list[-1] in GS_ALIASES + LOCAL_ALIASES:
        image_type = path_list.pop(-1)

    # Now for the tricky part. We don't actually know at this point if the rest
    # of the path is just a board | version (like R33-2341.0.0) or just a board
    # or just a version. So we do our best to do the right thing.
    board = default_board
    version = default_version or LATEST
    if len(path_list) == 1:
        path = path_list.pop(0)
        if path in (board, version):
            pass
        # Treat this as a version if it's one we know (contains default or
        # latest), or we were given an actual default board.
        elif default_version in path or LATEST in path or board is not None:
            version = path
        else:
            board = path

    elif len(path_list) == 2:
        # Assumes board/version.
        board = path_list.pop(0)
        version = path_list.pop(0)

    if path_list:
        raise XBuddyException(
            "Path is not valid. Could not figure out how to "
            "parse remaining components: %s." % path_list
        )

    logging.debug(
        "Get artifact '%s' with board %s and version %s'. Locally? %s",
        image_type,
        board,
        version,
        is_local,
    )

    return XBuddyComponents(image_type, board, version, is_local)


def parse(path: str, strict: bool = False) -> XBuddyComponents:
    """Given a fully formed URI, parse and return the XBuddy URI components.

    Based on observations, here's how the URI may be arranged, and what defaults
    are used if URI components aren't provided:

    Full URI: xbuddy://${LOCATION}/${BOARD}/${VERSION}/${IMAGE_TYPE}
        - ${LOCATION} specifies whether to check image folder locally or to
          retrieve an image from a remote host (i.e. a Google Storage bucket.)
          - DEFAULT: local
        - ${BOARD} specifies a particular build target; useful for testing
          hardware and/or overlay-specific software.
          - DEFAULT: None
        - ${VERSION} specifies a milestone-version identifier to target a
          specific ChromeOS release build.
          - DEFAULT: latest (an alias which gets resolved to the most recent
            milestone-version pair available).
        - ${IMAGE_TYPE} specifies the image type used to build an image in
          question. Different image types include useful tools for developers
          (e.g. extra packages used for testing software and hardware
          functionality). The most common targeted types are dev, test, and
          base.
            - DEFAULT:
                - If ${LOCATION} == 'local': ANY (an alias which resolves to any
                  usable image type.
                - If ${LOCATION} == 'remote': 'test'

    Any or all of these URI components may be omitted from the URI, but when
    any components are present, this hierarchical ordering should be respected.
    If reordered, the URI will not be parsed correctly.

    Args:
        path: The XBuddy URI. This method will perform URI validation and
            parsing, and the client is not expected perform any validation.
        strict: A client-specified option that enforces a requirement that the
            XBuddy URI begin with a well-formed scheme (e.g. 'xbuddy://').

    Returns:
        A Components tuple containing the parsed elements of the URI.

    Raises:
        XBuddyInvalidSchemeException if the scheme is not xbuddy, or is missing
        when strict checking is enforced.
    """
    parsed = urllib.parse.urlparse(path)

    if parsed.scheme == "":
        if strict:
            # Client has requested the scheme checking to be strict and require
            # the URI to start with 'xbuddy://'. Enforce this and raise an
            # exception.
            raise XBuddyInvalidSchemeException(
                "A URI scheme starting with 'xbuddy://' was expected but none "
                "was found."
            )
        else:
            logging.debug(
                'Assuming "%s" is an xbuddy path; scheme enforcement was not '
                "requested",
                path,
            )
    elif parsed.scheme != "xbuddy":
        raise XBuddyInvalidSchemeException(
            "Scheme of XBuddy URI was invalid. Provided path: %s" % path
        )

    return InterpretPath(parsed.netloc + parsed.path)