run_isolated.py: install CIPD packages

client/cipd.py

Index: client/cipd.py
diff --git a/client/cipd.py b/client/cipd.py
new file mode 100644
index 0000000000000000000000000000000000000000..39b5edf1a1f85b2ec66068fdc17cf505e710d0fb
--- /dev/null
+++ b/client/cipd.py
@@ -0,0 +1,403 @@
+# 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."""
+
+__version__ = '0.1'
+
+import hashlib
+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 ''
+
+
+class Error(Exception):
+  """Raised on CIPD errors."""
+
+
+def add_cipd_options(parser):
+  group = optparse.OptionGroup(parser, 'CIPD')
+  group.add_option(
+      '--cipd-server',
+      help='URL of the CIPD server. Only relevant with --cipd-package. '
+           'Default: "%default"',
+      default='https://chrome-infra-packages.appspot.com')

[M-A Ruel, 2016/06/07 20:43:39]

Please remove. We do not reference the chromium instances by default. The
current exception is luci-config and I'd like it to see go away too (be
configurable, without a default value).

[nodir, 2016/06/07 21:51:37]

Done.

+  group.add_option(
+      '--cipd-client-package',
+      help='Package of CIPD client. See --cipd-package for format. '
+           'Only relevant with --cipd-package. '
+           'Default: "%default"',
+      default='infra/tools/cipd/${platform}:latest')
+  group.add_option(
+      '--cipd-package',
+      help='CIPD package to install. '
+           'Format: "<package name template>:<version>". '
+           'Package name template is a CIPD package name with optional '
+           '${platform} and/or ${os_ver} parameters. '
+           '${platform} will be expanded to "<os>-<architecture>" and '
+           '${os_ver} will be expanded to OS version name. '
+           'This option can be specified more than once.',
+      action='append')
+  group.add_option(
+      '--cipd-cache',
+      help='CIPD cache directory, separate from isolate cache. '
+           'Only relevant with --cipd-package. '
+           'Default: "%default".')

[M-A Ruel, 2016/06/07 20:43:39]

there's no defaul?

[nodir, 2016/06/07 21:51:37]

https://cs.chromium.org/chromium/infra/luci/client/isolateserver.py?q=isolate...
does not have default either

defaults for both options are set by the caller. 

I've set it to '' so having %default does not hurt (it will display "" if not
set)

+  parser.add_option_group(group)
+
+
+def validate_cipd_options(parser, options):
+  """Calls parser.error on first found error among cipd options."""
+  if not options.cipd_package:
+    return
+  for p in options.cipd_package:
+    try:
+      parse_package(p)
+    except ValueError as ex:
+      parser.error('Invalid cipd package %r: %s' % (p, ex))
+
+  if not options.cipd_server:
+    parser.error('--cipd-package requires non-empty --cipd-server')
+
+  if not options.cipd_client_package:
+    parser.error('--cipd-package requires non-empty --cipd-client-package')
+  try:
+    parse_package(options.cipd_client_package)
+  except ValueError as ex:
+    parser.error(
+        'Invalid cipd client package %r: %s' %
+        (options.cipd_client_package, ex))
+
+
+class CipdClient(object):
+  """Installs packages."""
+
+  def __init__(self, binary_path, service_url=None):
+    """Initializes CipdClient.
+
+    Args:
+      binary_path (str): path to the CIPD client binary.
+      service_url (str): if not None, URL of the CIPD backend that overrides
+        the default one.
+    """
+    self.binary_path = binary_path
+    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 (str): list of packages to install, parsable by parse_pacakge().
+      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.
+
+    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)
+
+    list_file_handle, list_file_path = tempfile.mkstemp(
+        dir=tmp_dir, prefix=u'cipd-ensure-list-', suffix='.txt')
+    list_file_closed = False
+    try:
+      for p in packages:
+        pkg, version = parse_package(p)
+        pkg = render_package_name_template(pkg)
+        os.write(list_file_handle, '%s %s\n' % (pkg, version))
+      os.close(list_file_handle)
+      list_file_closed = True
+
+      cmd = [
+        self.binary_path, 'ensure',
+        '-root', site_root,
+        '-list', list_file_path,
+        '-verbose',  # this is safe because cipd-ensure does not print a lot
+      ]
+      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)))
+    finally:
+      if not list_file_closed:
+        os.close(list_file_handle)
+      fs.remove(list_file_path)
+
+
+def get_platform():
+  """Returns ${platform} parameter value.
+
+  Borrowed from
+  https://chromium.googlesource.com/infra/infra/+/aaf9586/build/build.py#204
+  """
+  # linux, mac or windows.
+  platform_variant = {
+    'darwin': 'mac',
+    'linux2': 'linux',
+    'win32': 'windows',
+  }.get(sys.platform)
+  if not platform_variant:
+    raise Error('Unknown OS: %s' % sys.platform)
+
+  # amd64, 386, etc.
+  platform_arch = {
+    'amd64': 'amd64',
+    'i386': '386',
+    'i686': '386',
+    'x86': '386',
+    'x86_64': 'amd64',
+  }.get(platform.machine().lower())
+  if not platform_arch:
+    platform_arch = 'amd64' if sys.maxsize > 2**32 else '386'

[M-A Ruel, 2016/06/07 20:43:39]

that's still highly incorrect on arm. amd64 and 386 are odd choices of keywords
(but I guess it's too late to change).

[Vadim Sh., 2016/06/07 20:59:42]

That's what Golang is uses :) I adopted it for cipd too. https://golang.org/dl/

For arm they use armv6l (I guess it's the minimal instruction set they need).

[nodir, 2016/06/07 21:51:37]

Done. Adapted from 
https://cs.chromium.org/chromium/infra/luci/appengine/swarming/swarming_bot/a...
it is not late because we actually never created cipd packages for arm

+  return '%s-%s' % (platform_variant, platform_arch)
+
+
+def get_os_ver():

[M-A Ruel, 2016/06/07 20:43:39]

I wonder what's the value for Windows and OSX, especially that the way the
string is created for Windows is useless in practice.

[nodir, 2016/06/07 21:51:37]

See examples below
Ask Vadim. I don't know what is the value of this string, but i added it for
compatibility with
https://chromium.googlesource.com/infra/infra/+/aaf9586/build/README.md

+  """Returns ${os_ver} parameter value.
+
+  Examples: 'ubuntu14_04' or 'mac10_9' or 'win6_1'.
+
+  Borrowed from
+  https://chromium.googlesource.com/infra/infra/+/aaf9586/build/build.py#204
+  """
+  if sys.platform == 'darwin':
+    # platform.mac_ver()[0] is '10.9.5'.
+    dist = platform.mac_ver()[0].split('.')
+    return 'mac%s_%s' % (dist[0], dist[1])
+
+  if sys.platform == 'linux2':
+    # platform.linux_distribution() is ('Ubuntu', '14.04', ...).
+    dist = platform.linux_distribution()
+    return '%s%s' % (dist[0].lower(), dist[1].replace('.', '_'))
+
+  if sys.platform == 'win32':
+    # platform.version() is '6.1.7601'.
+    dist = platform.version().split('.')
+    return 'win%s_%s' % (dist[0], dist[1])
+  raise Error('Unknown OS: %s' % sys.platform)
+
+
+def render_package_name_template(template):
+  """Expands template variables in a CIPD package name template."""
+  return (template
+      .replace('${platform}', get_platform())
+      .replace('${os_ver}', get_os_ver()))
+
+
+def parse_package(package):
+  """Parses a package in --cipd-package format.
+
+  Returns:
+    (package_name_template, version) tuple.
+
+  Raises:
+    ValueError if package name or version is not specified.
+  """
+  if not package:
+    raise ValueError('package is not specified')
+  parts = package.split(':', 1)
+  if len(parts) != 2:
+    raise ValueError('version is not specified')
+  return tuple(parts)
+
+
+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.
+  package_name = render_package_name_template(package_name)
+  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.
+
+  Returns:
+    Path to the fetched file inside disk_cache.
+
+  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 disk_cache.item_path(instance_id)
+    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')
+
+
+def get_client(
+      service_url, package_name, version, cache_dir, timeout=None):
+  """Creates a CipdClient. A blocking call.
+
+  Args:
+      service_url (str): URL of the CIPD backend.
+      package_name (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.
+
+  Returns:
+    CipdClient.
+
+  Raises:
+    Error if CIPD client version cannot be resolved or client cannot be fetched.
+  """
+  timeoutfn = tools.sliding_timeout(timeout)
+
+  package_name = render_package_name_template(package_name)
+
+  # 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
+  else:
+    # version_cache is {version_digest -> 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)
+
+    # Convert |version| to a string that may be used as a filename in disk cache
+    # by hashing it.
+    version_digest = hashlib.sha1(version).hexdigest()
+    try:
+      instance_id = version_cache.read(version_digest)
+    except isolateserver.CacheMiss:
+      instance_id = resolve_version(
+          service_url, package_name, version, timeout=timeoutfn())
+      version_cache.write(version_digest, instance_id)
+
+  # instance_cache is {instance_id -> client binary} mapping.
+  # It is bounded by 5 client versions.
+  instance_cache = isolateserver.DiskCache(

[M-A Ruel, 2016/06/07 20:43:39]

The instance should outlive the mapped files

[nodir, 2016/06/07 21:51:37]

I don't understand this comment. what is the difference between "instance" and
"mapped files"? why wouldn't it outlive if we have exclusive access to the
cache?

[M-A Ruel, 2016/06/08 20:37:39]

instance = object. This means instance_cache should be created by the caller.

you should use
with version_cache:

so it activates its initialization code. Right now it's not doing much but it
could later.

[nodir, 2016/06/08 22:35:32]

Made get_client return a context manager that yields the client.
 
Added with-clause for version_cache

+      unicode(os.path.join(cache_dir, 'clients')),
+      isolateserver.CachePolicies(0, 0, 5),
+      hashlib.sha1)
+  # Get the path to the client binary inside instance cache.
+  try:
+    cached_binary_path = instance_cache.item_path(instance_id)
+  except isolateserver.CacheMiss:
+    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())
+    cached_binary_path = _fetch_cipd_client(
+        instance_cache, instance_id, fetch_url, timeoutfn)
+
+  # A single bot can run multiple swarming clients, but ATM they do not share

[M-A Ruel, 2016/06/07 20:43:39]

s/single bot/single host/
s/clients/bots/
s/share cache/share same root bot directory/

[nodir, 2016/06/07 21:51:37]

Done.

+  # cache. Thus, it is safe to use the same name for the binary.
+  binary_path = unicode(os.path.join(cache_dir, 'cipd' + EXECUTABLE_SUFFIX))
+  if fs.isfile(binary_path):
+    fs.unlink(binary_path)
+  file_path.hardlink(cached_binary_path, binary_path)

[M-A Ruel, 2016/06/07 20:43:39]

I'd prefer to use instance_cache.hardlink()

[nodir, 2016/06/07 21:51:37]

yeah, that's better. I thought hardlink allows to _put_ a file into the cache
using a hardlink, not vice versa

+  fs.chmod(binary_path, 0711)  # -rwx--x--x

[M-A Ruel, 2016/06/07 20:43:39]

Why write bit set?

[nodir, 2016/06/07 21:51:37]

I assumed you cannot delete a file without w bit. Turns out it is false. Removed
write bit

[M-A Ruel, 2016/06/08 20:37:39]

In practice, that is not true on Windows; as removing the w bit *add* the
read_only bit on Windows. And unlike on POSIX, where the directory ACL controls
which file can be deleted, on Windows it's the file's ACL that decide if the
file can be deleted.

[nodir, 2016/06/08 22:35:32]

I've sshed to a windows machine, created a file with 0111 permissions, verified
that oct(os.stat('the file').st_mode & 0777) == '0111' and then ran os.remove on
it. it was deleted successfully, so current code is probably fine

+
+  return CipdClient(binary_path)