| # Copyright 2016 The LUCI Authors. All rights reserved. |
| # Use of this source code is governed under the Apache License, Version 2.0 |
| # that can be found in the LICENSE file. |
| |
| """Fetches CIPD client and installs packages.""" |
| |
| import contextlib |
| import hashlib |
| import json |
| import logging |
| import optparse |
| import os |
| import platform |
| import sys |
| import tempfile |
| import time |
| import urllib |
| |
| from utils import file_path |
| from utils import fs |
| from utils import net |
| from utils import subprocess42 |
| from utils import tools |
| import isolated_format |
| import isolateserver |
| |
| |
| # .exe on Windows. |
| EXECUTABLE_SUFFIX = '.exe' if sys.platform == 'win32' else '' |
| |
| |
| if sys.platform == 'win32': |
| def _ensure_batfile(client_path): |
| base, _ = os.path.splitext(client_path) |
| with open(base+".bat", 'w') as f: |
| f.write('\n'.join([ # python turns \n into CRLF |
| '@set CIPD="%~dp0cipd.exe"', |
| '@shift', |
| '@%CIPD% %*' |
| ])) |
| else: |
| def _ensure_batfile(_client_path): |
| pass |
| |
| |
| class Error(Exception): |
| """Raised on CIPD errors.""" |
| |
| |
| def add_cipd_options(parser): |
| group = optparse.OptionGroup(parser, 'CIPD') |
| group.add_option( |
| '--cipd-enabled', |
| help='Enable CIPD client bootstrap. Implied by --cipd-package.', |
| action='store_true', |
| default=False) |
| group.add_option( |
| '--cipd-server', |
| help='URL of the CIPD server. ' |
| 'Only relevant with --cipd-enabled or --cipd-package.') |
| group.add_option( |
| '--cipd-client-package', |
| help='Package name of CIPD client with optional parameters described in ' |
| '--cipd-package help. ' |
| 'Only relevant with --cipd-enabled or --cipd-package. ' |
| 'Default: "%default"', |
| default='infra/tools/cipd/${platform}') |
| group.add_option( |
| '--cipd-client-version', |
| help='Version of CIPD client. ' |
| 'Only relevant with --cipd-enabled or --cipd-package. ' |
| 'Default: "%default"', |
| default='latest') |
| group.add_option( |
| '--cipd-package', |
| dest='cipd_packages', |
| help='A CIPD package to install. ' |
| 'Format is "<path>:<package_name>:<version>". ' |
| '"path" is installation directory relative to run_dir, ' |
| 'defaults to ".". ' |
| '"package_name" may have ${platform} parameter: it will be ' |
| 'expanded to "<os>-<architecture>". ' |
| 'The option can be specified multiple times.', |
| action='append', |
| default=[]) |
| group.add_option( |
| '--cipd-cache', |
| help='CIPD cache directory, separate from isolate cache. ' |
| 'Only relevant with --cipd-enabled or --cipd-package. ' |
| 'Default: "%default".', |
| default='') |
| parser.add_option_group(group) |
| |
| |
| def validate_cipd_options(parser, options): |
| """Calls parser.error on first found error among cipd options.""" |
| if options.cipd_packages: |
| options.cipd_enabled = True |
| |
| if not options.cipd_enabled: |
| return |
| |
| for pkg in options.cipd_packages: |
| parts = pkg.split(':', 2) |
| if len(parts) != 3: |
| parser.error('invalid package "%s": must have at least 2 colons' % pkg) |
| _path, name, version = parts |
| if not name: |
| parser.error('invalid package "%s": package name is not specified' % pkg) |
| if not version: |
| parser.error('invalid package "%s": version is not specified' % pkg) |
| |
| if not options.cipd_server: |
| parser.error('cipd is enabled, --cipd-server is required') |
| |
| if not options.cipd_client_package: |
| parser.error( |
| 'cipd is enabled, --cipd-client-package is required') |
| if not options.cipd_client_version: |
| parser.error( |
| 'cipd is enabled, --cipd-client-version is required') |
| |
| |
| class CipdClient(object): |
| """Installs packages.""" |
| |
| def __init__(self, binary_path, package_name, instance_id, service_url): |
| """Initializes CipdClient. |
| |
| Args: |
| binary_path (str): path to the CIPD client binary. |
| package_name (str): the CIPD package name for the client itself. |
| instance_id (str): the CIPD instance_id for the client itself. |
| service_url (str): if not None, URL of the CIPD backend that overrides |
| the default one. |
| """ |
| self.binary_path = binary_path |
| self.package_name = package_name |
| self.instance_id = instance_id |
| self.service_url = service_url |
| |
| def ensure( |
| self, site_root, packages, cache_dir=None, tmp_dir=None, timeout=None): |
| """Ensures that packages installed in |site_root| equals |packages| set. |
| |
| Blocking call. |
| |
| Args: |
| site_root (str): where to install packages. |
| packages: dict of subdir -> list of (package_template, version) tuples. |
| cache_dir (str): if set, cache dir for cipd binary own cache. |
| Typically contains packages and tags. |
| tmp_dir (str): if not None, dir for temp files. |
| timeout (int): if not None, timeout in seconds for this function to run. |
| |
| Returns: |
| Pinned packages in the form of {subdir: [(package_name, package_id)]}, |
| which correspond 1:1 with the input packages argument. |
| |
| Raises: |
| Error if could not install packages or timed out. |
| """ |
| timeoutfn = tools.sliding_timeout(timeout) |
| logging.info('Installing packages %r into %s', packages, site_root) |
| |
| ensure_file_handle, ensure_file_path = tempfile.mkstemp( |
| dir=tmp_dir, prefix=u'cipd-ensure-file-', suffix='.txt') |
| json_out_file_handle, json_file_path = tempfile.mkstemp( |
| dir=tmp_dir, prefix=u'cipd-ensure-result-', suffix='.json') |
| os.close(json_out_file_handle) |
| |
| try: |
| try: |
| for subdir, pkgs in sorted(packages.iteritems()): |
| if '\n' in subdir: |
| raise Error( |
| 'Could not install packages; subdir %r contains newline' % subdir) |
| os.write(ensure_file_handle, '@Subdir %s\n' % (subdir,)) |
| for pkg, version in pkgs: |
| os.write(ensure_file_handle, '%s %s\n' % (pkg, version)) |
| finally: |
| os.close(ensure_file_handle) |
| |
| cmd = [ |
| self.binary_path, 'ensure', |
| '-root', site_root, |
| '-ensure-file', ensure_file_path, |
| '-verbose', # this is safe because cipd-ensure does not print a lot |
| '-json-output', json_file_path, |
| ] |
| if cache_dir: |
| cmd += ['-cache-dir', cache_dir] |
| if self.service_url: |
| cmd += ['-service-url', self.service_url] |
| |
| logging.debug('Running %r', cmd) |
| process = subprocess42.Popen( |
| cmd, stdout=subprocess42.PIPE, stderr=subprocess42.PIPE) |
| output = [] |
| for pipe_name, line in process.yield_any_line(timeout=0.1): |
| to = timeoutfn() |
| if to is not None and to <= 0: |
| raise Error( |
| 'Could not install packages; took more than %d seconds' % timeout) |
| if not pipe_name: |
| # stdout or stderr was closed, but yield_any_line still may have |
| # something to yield. |
| continue |
| output.append(line) |
| if pipe_name == 'stderr': |
| logging.debug('cipd client: %s', line) |
| else: |
| logging.info('cipd client: %s', line) |
| |
| exit_code = process.wait(timeout=timeoutfn()) |
| if exit_code != 0: |
| raise Error( |
| 'Could not install packages; exit code %d\noutput:%s' % ( |
| exit_code, '\n'.join(output))) |
| with open(json_file_path) as jfile: |
| result_json = json.load(jfile) |
| return { |
| subdir: [(x['package'], x['instance_id']) for x in pins] |
| for subdir, pins in result_json['result'].iteritems() |
| } |
| finally: |
| fs.remove(ensure_file_path) |
| fs.remove(json_file_path) |
| |
| |
| def get_platform(): |
| """Returns ${platform} parameter value. |
| |
| The logic is similar to |
| https://chromium.googlesource.com/chromium/tools/build/+/6c5c7e9c/scripts/slave/infra_platform.py |
| """ |
| # linux, mac or windows. |
| os_name = { |
| 'darwin': 'mac', |
| 'linux2': 'linux', |
| 'win32': 'windows', |
| }.get(sys.platform) |
| if not os_name: |
| raise Error('Unknown OS: %s' % sys.platform) |
| |
| # Normalize machine architecture. Some architectures are identical or |
| # compatible with others. We collapse them into one. |
| arch = platform.machine().lower() |
| if arch in ('arm64', 'aarch64'): |
| arch = 'arm64' |
| elif arch.startswith('armv') and arch.endswith('l'): |
| # 32-bit ARM: Standardize on ARM v6 baseline. |
| arch = 'armv6l' |
| elif arch in ('amd64', 'x86_64'): |
| arch = 'amd64' |
| elif arch in ('i386', 'i686', 'x86'): |
| arch = '386' |
| |
| # If using a 32-bit python on x86_64 kernel on Linux, "downgrade" the arch to |
| # 32-bit too (this is the bitness of the userland). |
| python_bits = 64 if sys.maxsize > 2**32 else 32 |
| if os_name == 'linux' and arch == 'amd64' and python_bits == 32: |
| arch = '386' |
| |
| return '%s-%s' % (os_name, arch) |
| |
| |
| def _check_response(res, fmt, *args): |
| """Raises Error if response is bad.""" |
| if not res: |
| raise Error('%s: no response' % (fmt % args)) |
| |
| if res.get('status') != 'SUCCESS': |
| raise Error('%s: %s' % ( |
| fmt % args, |
| res.get('error_message') or 'status is %s' % res.get('status'))) |
| |
| |
| def resolve_version(cipd_server, package_name, version, timeout=None): |
| """Resolves a package instance version (e.g. a tag) to an instance id.""" |
| url = '%s/_ah/api/repo/v1/instance/resolve?%s' % ( |
| cipd_server, |
| urllib.urlencode({ |
| 'package_name': package_name, |
| 'version': version, |
| })) |
| res = net.url_read_json(url, timeout=timeout) |
| _check_response(res, 'Could not resolve version %s:%s', package_name, version) |
| instance_id = res.get('instance_id') |
| if not instance_id: |
| raise Error('Invalid resolveVersion response: no instance id') |
| return instance_id |
| |
| |
| def get_client_fetch_url(service_url, package_name, instance_id, timeout=None): |
| """Returns a fetch URL of CIPD client binary contents. |
| |
| Raises: |
| Error if cannot retrieve fetch URL. |
| """ |
| # Fetch the URL of the binary from CIPD backend. |
| url = '%s/_ah/api/repo/v1/client?%s' % (service_url, urllib.urlencode({ |
| 'package_name': package_name, |
| 'instance_id': instance_id, |
| })) |
| res = net.url_read_json(url, timeout=timeout) |
| _check_response( |
| res, 'Could not fetch CIPD client %s:%s',package_name, instance_id) |
| fetch_url = res.get('client_binary', {}).get('fetch_url') |
| if not fetch_url: |
| raise Error('Invalid fetchClientBinary response: no fetch_url') |
| return fetch_url |
| |
| |
| def _fetch_cipd_client(disk_cache, instance_id, fetch_url, timeoutfn): |
| """Fetches cipd binary to |disk_cache|. |
| |
| Retries requests with exponential back-off. |
| |
| Raises: |
| Error if could not fetch content. |
| """ |
| sleep_time = 1 |
| for attempt in xrange(5): |
| if attempt > 0: |
| if timeoutfn() is not None and timeoutfn() < sleep_time: |
| raise Error('Could not fetch CIPD client: timeout') |
| logging.warning('Will retry to fetch CIPD client in %ds', sleep_time) |
| time.sleep(sleep_time) |
| sleep_time *= 2 |
| |
| try: |
| res = net.url_open(fetch_url, timeout=timeoutfn()) |
| if res: |
| disk_cache.write(instance_id, res.iter_content(64 * 1024)) |
| return |
| except net.TimeoutError as ex: |
| raise Error('Could not fetch CIPD client: %s', ex) |
| except net.NetError as ex: |
| logging.warning( |
| 'Could not fetch CIPD client on attempt #%d: %s', attempt + 1, ex) |
| |
| raise Error('Could not fetch CIPD client after 5 retries') |
| |
| |
| @contextlib.contextmanager |
| def get_client(service_url, package_template, version, cache_dir, timeout=None): |
| """Returns a context manager that yields a CipdClient. A blocking call. |
| |
| Upon exit from the context manager, the client binary may be deleted |
| (if the internal cache is full). |
| |
| Args: |
| service_url (str): URL of the CIPD backend. |
| package_template (str): package name template of the CIPD client. |
| version (str): version of CIPD client package. |
| cache_dir: directory to store instance cache, version cache |
| and a hardlink to the client binary. |
| timeout (int): if not None, timeout in seconds for this function. |
| |
| Yields: |
| CipdClient. |
| |
| Raises: |
| Error if CIPD client version cannot be resolved or client cannot be fetched. |
| """ |
| timeoutfn = tools.sliding_timeout(timeout) |
| |
| # Package names are always lower case. |
| # TODO(maruel): Assert instead? |
| package_name = package_template.lower().replace('${platform}', get_platform()) |
| |
| # Resolve version to instance id. |
| # Is it an instance id already? They look like HEX SHA1. |
| if isolated_format.is_valid_hash(version, hashlib.sha1): |
| instance_id = version |
| elif ':' in version: # it's an immutable tag, cache the resolved version |
| # version_cache is {hash(package_name, tag) -> instance id} mapping. |
| # It does not take a lot of disk space. |
| version_cache = isolateserver.DiskCache( |
| unicode(os.path.join(cache_dir, 'versions')), |
| isolateserver.CachePolicies(0, 0, 300), |
| hashlib.sha1, |
| trim=True) |
| with version_cache: |
| version_cache.cleanup() |
| # Convert (package_name, version) to a string that may be used as a |
| # filename in disk cache by hashing it. |
| version_digest = hashlib.sha1( |
| '%s\n%s' % (package_name, version)).hexdigest() |
| try: |
| with version_cache.getfileobj(version_digest) as f: |
| instance_id = f.read() |
| except isolateserver.CacheMiss: |
| instance_id = resolve_version( |
| service_url, package_name, version, timeout=timeoutfn()) |
| version_cache.write(version_digest, instance_id) |
| else: # it's a ref, hit the backend |
| instance_id = resolve_version( |
| service_url, package_name, version, timeout=timeoutfn()) |
| |
| # instance_cache is {instance_id -> client binary} mapping. |
| # It is bounded by 5 client versions. |
| instance_cache = isolateserver.DiskCache( |
| unicode(os.path.join(cache_dir, 'clients')), |
| isolateserver.CachePolicies(0, 0, 5), |
| hashlib.sha1, |
| trim=True) |
| with instance_cache: |
| instance_cache.cleanup() |
| if instance_id not in instance_cache: |
| logging.info('Fetching CIPD client %s:%s', package_name, instance_id) |
| fetch_url = get_client_fetch_url( |
| service_url, package_name, instance_id, timeout=timeoutfn()) |
| _fetch_cipd_client(instance_cache, instance_id, fetch_url, timeoutfn) |
| |
| # A single host can run multiple swarming bots, but ATM they do not share |
| # same root bot directory. Thus, it is safe to use the same name for the |
| # binary. |
| cipd_bin_dir = unicode(os.path.join(cache_dir, 'bin')) |
| binary_path = os.path.join(cipd_bin_dir, 'cipd' + EXECUTABLE_SUFFIX) |
| if fs.isfile(binary_path): |
| file_path.remove(binary_path) |
| else: |
| file_path.ensure_tree(cipd_bin_dir) |
| |
| with instance_cache.getfileobj(instance_id) as f: |
| isolateserver.putfile(f, binary_path, 0511) # -r-x--x--x |
| |
| _ensure_batfile(binary_path) |
| |
| yield CipdClient( |
| binary_path, |
| package_name=package_name, |
| instance_id=instance_id, |
| service_url=service_url) |
| |
| |
| def parse_package_args(packages): |
| """Parses --cipd-package arguments. |
| |
| Assumes |packages| were validated by validate_cipd_options. |
| |
| Returns: |
| A list of [(path, package_name, version), ...] |
| """ |
| result = [] |
| for pkg in packages: |
| path, name, version = pkg.split(':', 2) |
| if not name: |
| raise Error('Invalid package "%s": package name is not specified' % pkg) |
| if not version: |
| raise Error('Invalid package "%s": version is not specified' % pkg) |
| result.append((path, name, version)) |
| return result |
