blob: 33cd60651f86194d036e452d4b0b5bd3d0bcecdc [file] [log] [blame]
# -*- coding: utf-8 -*-
# Copyright 2019 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""The Binhost API interacts with Portage binhosts and Packages files."""
from __future__ import print_function
import functools
import os
from chromite.lib import binpkg
from chromite.lib import constants
from chromite.lib import cros_build_lib
from chromite.lib import osutils
from chromite.lib import parallel
from chromite.lib import portage_util
from chromite.utils import key_value_store
# The name of the ACL argument file.
_GOOGLESTORAGE_GSUTIL_FILE = 'googlestorage_acl.txt'
# The name of the package file (relative to sysroot) where the list of packages
# for dev-install is stored.
_DEV_INSTALL_PACKAGES_FILE = 'build/dev-install/package.installable'
class Error(Exception):
"""Base error class for the module."""
class EmptyPrebuiltsRoot(Error):
"""When a prebuilts root is unexpectedly empty."""
class NoAclFileFound(Error):
"""No ACL file could be found."""
def _ValidateBinhostConf(path, key):
"""Validates the binhost conf file defines only one environment variable.
This function is effectively a sanity check that ensures unexpected
configuration is not clobbered by conf overwrites.
path: Path to the file to validate.
key: Expected binhost key.
ValueError: If file defines != 1 environment variable.
if not os.path.exists(path):
# If the conf file does not exist, e.g. with new targets, then whatever.
kvs = key_value_store.LoadFile(path)
if not kvs:
raise ValueError(
'Found empty .conf file %s when a non-empty one was expected.' % path)
elif len(kvs) > 1:
raise ValueError(
'Conf file %s must define exactly 1 variable. '
'Instead found: %r' % (path, kvs))
elif key not in kvs:
raise KeyError('Did not find key %s in %s' % (key, path))
def _ValidatePrebuiltsFiles(prebuilts_root, prebuilts_paths):
"""Validate all prebuilt files exist.
prebuilts_root: Absolute path to root directory containing prebuilts.
prebuilts_paths: List of file paths relative to root, to be verified.
LookupError: If any prebuilt archive does not exist.
for prebuilt_path in prebuilts_paths:
full_path = os.path.join(prebuilts_root, prebuilt_path)
if not os.path.exists(full_path):
raise LookupError('Prebuilt archive %s does not exist' % full_path)
def _ValidatePrebuiltsRoot(target, prebuilts_root):
"""Validate the given prebuilts root exists.
If the root does not exist, it probably means the build target did not build
successfully, so warn callers appropriately.
target: The build target in question.
prebuilts_root: The expected root directory for the target's prebuilts.
EmptyPrebuiltsRoot: If prebuilts root does not exist.
if not os.path.exists(prebuilts_root):
raise EmptyPrebuiltsRoot(
'Expected to find prebuilts for build target %s at %s. '
'Did %s build successfully?' % (target, prebuilts_root, target))
def GetPrebuiltsRoot(chroot, sysroot, build_target):
"""Find the root directory with binary prebuilts for the given sysroot.
chroot (chroot_lib.Chroot): The chroot where the sysroot lives.
sysroot (sysroot_lib.Sysroot): The sysroot.
build_target (build_target_lib.BuildTarget): The build target.
Absolute path to the root directory with the target's prebuilt archives.
root = os.path.join(chroot.path, sysroot.path.lstrip(os.sep), 'packages')
_ValidatePrebuiltsRoot(build_target, root)
return root
def GetPrebuiltsFiles(prebuilts_root):
"""Find paths to prebuilts at the given root directory.
Assumes the root contains a Portage package index named Packages.
The package index paths are used to de-duplicate prebuilts uploaded. The
immediate consequence of this is reduced storage usage. The non-obvious
consequence is the shared packages generally end up with public permissions,
while the board-specific packages end up with private permissions. This is
what is meant to happen, but a further consequence of that is that when
something happens that causes the toolchains to be uploaded as a private
board's package, the board will not be able to build properly because
it won't be able to fetch the toolchain packages, because they're expected
to be public.
prebuilts_root: Absolute path to root directory containing a package index.
List of paths to all prebuilt archives, relative to the root.
package_index = binpkg.GrabLocalPackageIndex(prebuilts_root)
prebuilt_paths = []
for package in package_index.packages:
prebuilt_paths.append(package['CPV'] + '.tbz2')
include_debug_symbols = package.get('DEBUG_SYMBOLS')
if cros_build_lib.BooleanShellValue(include_debug_symbols, default=False):
prebuilt_paths.append(package['CPV'] + '.debug.tbz2')
_ValidatePrebuiltsFiles(prebuilts_root, prebuilt_paths)
return prebuilt_paths
def UpdatePackageIndex(prebuilts_root, upload_uri, upload_path, sudo=False):
"""Update package index with information about where it will be uploaded.
This causes the existing Packages file to be overwritten.
prebuilts_root: Absolute path to root directory containing binary prebuilts.
upload_uri: The URI (typically GS bucket) where prebuilts will be uploaded.
upload_path: The path at the URI for the prebuilts.
sudo (bool): Whether to write the file as the root user.
Path to the new Package index.
assert not upload_path.startswith('/')
package_index = binpkg.GrabLocalPackageIndex(prebuilts_root)
package_index.SetUploadLocation(upload_uri, upload_path)
package_index.header['TTL'] = 60 * 60 * 24 * 365
package_index_path = os.path.join(prebuilts_root, 'Packages')
package_index.WriteFile(package_index_path, sudo=sudo)
return package_index_path
def SetBinhost(target, key, uri, private=True):
"""Set binhost configuration for the given build target.
A binhost is effectively a key (Portage env variable) pointing to a URL
that contains binaries. The configuration is set in .conf files at static
directories on a build target by build target (and host by host) basis.
This function updates the .conf file by completely rewriting it.
target: The build target to set configuration for.
key: The binhost key to set, e.g. POSTSUBMIT_BINHOST.
uri: The new value for the binhost key, e.g. gs://chromeos-prebuilt/foo/bar.
private: Whether or not the build target is private.
Path to the updated .conf file.
conf_root = os.path.join(
constants.PRIVATE_BINHOST_CONF_DIR if private else
constants.PUBLIC_BINHOST_CONF_DIR, 'target')
conf_file = '%s-%s.conf' % (target, key)
conf_path = os.path.join(conf_root, conf_file)
_ValidateBinhostConf(conf_path, key)
osutils.WriteFile(conf_path, '%s="%s"' % (key, uri))
return conf_path
def RegenBuildCache(chroot, overlay_type):
"""Regenerate the Build Cache for the given target.
chroot (chroot_lib): The chroot where the regen command will be run.
overlay_type: one of "private", "public", or "both".
list[str]: The overlays with updated caches.
overlays = portage_util.FindOverlays(overlay_type)
task = functools.partial(
portage_util.RegenCache, commit_changes=False, chroot=chroot)
task_inputs = [[o] for o in overlays if os.path.isdir(o)]
results = parallel.RunTasksInProcessPool(task, task_inputs)
# Filter out all of the unchanged-overlay results.
return [overlay_dir for overlay_dir in results if overlay_dir]
def GetPrebuiltAclArgs(build_target):
"""Read and parse the GS ACL file from the private overlays.
build_target (build_target_lib.BuildTarget): The build target.
list[list[str]]: A list containing all of the [arg, value] pairs. E.g.
[['-g', 'group_id:READ'], ['-u', 'user:FULL_CONTROL']]
acl_file = portage_util.FindOverlayFile(_GOOGLESTORAGE_GSUTIL_FILE,
if not acl_file:
raise NoAclFileFound('No ACL file found for %s.' %
lines = osutils.ReadFile(acl_file).splitlines()
# Remove comments.
lines = [line.split('#', 1)[0].strip() for line in lines]
# Remove empty lines.
lines = [line.strip() for line in lines if line.strip()]
return [line.split() for line in lines]
def GetBinhosts(build_target):
"""Get the binhosts for the build target.
build_target (build_target_lib.BuildTarget): The build target.
list[str]: The build target's binhosts.
binhosts = portage_util.PortageqEnvvar('PORTAGE_BINHOST',,
return binhosts.split() if binhosts else []
def ReadDevInstallPackageFile(filename):
"""Parse the dev-install package file.
filename (str): The full path to the dev-install package list.
list[str]: The packages in the package list file.
with open(filename) as f:
return [line.strip() for line in f]
def ReadDevInstallFilesToCreatePackageIndex(chroot, sysroot, package_index_path,
upload_uri, upload_path):
"""Create dev-install Package index specified by package_index_path
The current Packages file is read and a new Packages file is created based
on the subset of packages in the _DEV_INSTALL_PACKAGES_FILE.
chroot (chroot_lib.Chroot): The chroot where the sysroot lives.
sysroot (sysroot_lib.Sysroot): The sysroot.
package_index_path (str): Path to the Packages file to be created.
upload_uri: The URI (typically GS bucket) where prebuilts will be uploaded.
upload_path: The path at the URI for the prebuilts.
list[str]: The list of packages contained in package_index_path,
where each package string is a category/file.
# Read the dev-install binhost package file
devinstall_binhost_filename = chroot.full_path(sysroot.path,
devinstall_package_list = ReadDevInstallPackageFile(
# Read the Packages file, remove packages not in package_list
package_path = chroot.full_path(sysroot.path, 'packages')
CreateFilteredPackageIndex(package_path, devinstall_package_list,
upload_uri, upload_path)
# We have the list of packages, create full path and verify each one.
upload_targets_list = GetPrebuiltsForPackages(
package_path, devinstall_package_list)
return upload_targets_list
def CreateFilteredPackageIndex(package_path, devinstall_package_list,
upload_uri, upload_path, sudo=False):
"""Create Package file for dev-install process.
The created package file (package_index_path) contains only the
packages from the system packages file (in package_path) that are in the
devinstall_package_list. The new package file will use the provided values
for upload_uri and upload_path.
package_path (str): Absolute path to the standard Packages file.
devinstall_package_list (list[str]): Packages from packages.installable
package_index_path (str): Absolute path for new Packages file.
upload_uri (str): The URI where prebuilts will be uploaded.
upload_path (str): The path at the URI for the prebuilts.
sudo (bool): Whether to write the file as the root user.
def ShouldFilterPackage(package):
"""Local func to filter packages not in the devinstall_package_list
package (dict): Dictionary with key 'CPV' and package name as value
True (filter) if not in the devinstall_package_list, else False (don't
filter) if in the devinstall_package_list
value = package['CPV']
if value in devinstall_package_list:
return False
return True
package_index = binpkg.GrabLocalPackageIndex(package_path)
package_index.SetUploadLocation(upload_uri, upload_path)
package_index.header['TTL'] = 60 * 60 * 24 * 365
package_index.WriteFile(package_index_path, sudo=sudo)
def GetPrebuiltsForPackages(package_root, package_list):
"""Create list of file paths for the package list and validate they exist.
package_root (str): Path to 'packages' directory.
package_list (list[str]): List of packages.
List of validated targets.
upload_targets_list = []
for pkg in package_list:
zip_target = pkg + '.tbz2'
full_pkg_path = os.path.join(package_root, pkg) + '.tbz2'
if not os.path.exists(full_pkg_path):
raise LookupError('DevInstall archive %s does not exist' % full_pkg_path)
return upload_targets_list