src/tools/swarming_client/run_isolated.py

   1 #!/usr/bin/env python
   2 # Copyright (c) 2012 The Chromium Authors. All rights reserved.
   3 # Use of this source code is governed by a BSD-style license that can be
   4 # found in the LICENSE file.
   5
   6 """Reads a .isolated, creates a tree of hardlinks and runs the test.
   7
   8 Keeps a local cache.
   9 """
  10
  11 __version__ = '0.2'
  12
  13 import ctypes
  14 import logging
  15 import optparse
  16 import os
  17 import re
  18 import shutil
  19 import stat
  20 import subprocess
  21 import sys
  22 import tempfile
  23 import time
  24
  25 from third_party.depot_tools import fix_encoding
  26
  27 from utils import lru
  28 from utils import threading_utils
  29 from utils import tools
  30 from utils import zip_package
  31
  32 import isolateserver
  33
  34
  35 # Absolute path to this file (can be None if running from zip on Mac).
  36 THIS_FILE_PATH = os.path.abspath(__file__) if __file__ else None
  37
  38 # Directory that contains this file (might be inside zip package).
  39 BASE_DIR = os.path.dirname(THIS_FILE_PATH) if __file__ else None
  40
  41 # Directory that contains currently running script file.
  42 if zip_package.get_main_script_path():
  43   MAIN_DIR = os.path.dirname(
  44       os.path.abspath(zip_package.get_main_script_path()))
  45 else:
  46   # This happens when 'import run_isolated' is executed at the python
  47   # interactive prompt, in that case __file__ is undefined.
  48   MAIN_DIR = None
  49
  50 # Types of action accepted by link_file().
  51 HARDLINK, HARDLINK_WITH_FALLBACK, SYMLINK, COPY = range(1, 5)
  52
  53 # The name of the log file to use.
  54 RUN_ISOLATED_LOG_FILE = 'run_isolated.log'
  55
  56 # The name of the log to use for the run_test_cases.py command
  57 RUN_TEST_CASES_LOG = 'run_test_cases.log'
  58
  59
  60 # Used by get_flavor().
  61 FLAVOR_MAPPING = {
  62   'cygwin': 'win',
  63   'win32': 'win',
  64   'darwin': 'mac',
  65   'sunos5': 'solaris',
  66   'freebsd7': 'freebsd',
  67   'freebsd8': 'freebsd',
  68 }
  69
  70
  71 def get_as_zip_package(executable=True):
  72   """Returns ZipPackage with this module and all its dependencies.
  73
  74   If |executable| is True will store run_isolated.py as __main__.py so that
  75   zip package is directly executable be python.
  76   """
  77   # Building a zip package when running from another zip package is
  78   # unsupported and probably unneeded.
  79   assert not zip_package.is_zipped_module(sys.modules[__name__])
  80   assert THIS_FILE_PATH
  81   assert BASE_DIR
  82   package = zip_package.ZipPackage(root=BASE_DIR)
  83   package.add_python_file(THIS_FILE_PATH, '__main__.py' if executable else None)
  84   package.add_python_file(os.path.join(BASE_DIR, 'isolateserver.py'))
  85   package.add_directory(os.path.join(BASE_DIR, 'third_party'))
  86   package.add_directory(os.path.join(BASE_DIR, 'utils'))
  87   return package
  88
  89
  90 def get_flavor():
  91   """Returns the system default flavor. Copied from gyp/pylib/gyp/common.py."""
  92   return FLAVOR_MAPPING.get(sys.platform, 'linux')
  93
  94
  95 def os_link(source, link_name):
  96   """Add support for os.link() on Windows."""
  97   if sys.platform == 'win32':
  98     if not ctypes.windll.kernel32.CreateHardLinkW(
  99         unicode(link_name), unicode(source), 0):
 100       raise OSError()
 101   else:
 102     os.link(source, link_name)
 103
 104
 105 def readable_copy(outfile, infile):
 106   """Makes a copy of the file that is readable by everyone."""
 107   shutil.copy2(infile, outfile)
 108   read_enabled_mode = (os.stat(outfile).st_mode | stat.S_IRUSR |
 109                        stat.S_IRGRP | stat.S_IROTH)
 110   os.chmod(outfile, read_enabled_mode)
 111
 112
 113 def link_file(outfile, infile, action):
 114   """Links a file. The type of link depends on |action|."""
 115   logging.debug('Mapping %s to %s' % (infile, outfile))
 116   if action not in (HARDLINK, HARDLINK_WITH_FALLBACK, SYMLINK, COPY):
 117     raise ValueError('Unknown mapping action %s' % action)
 118   if not os.path.isfile(infile):
 119     raise isolateserver.MappingError('%s is missing' % infile)
 120   if os.path.isfile(outfile):
 121     raise isolateserver.MappingError(
 122         '%s already exist; insize:%d; outsize:%d' %
 123         (outfile, os.stat(infile).st_size, os.stat(outfile).st_size))
 124
 125   if action == COPY:
 126     readable_copy(outfile, infile)
 127   elif action == SYMLINK and sys.platform != 'win32':
 128     # On windows, symlink are converted to hardlink and fails over to copy.
 129     os.symlink(infile, outfile)  # pylint: disable=E1101
 130   else:
 131     try:
 132       os_link(infile, outfile)
 133     except OSError as e:
 134       if action == HARDLINK:
 135         raise isolateserver.MappingError(
 136             'Failed to hardlink %s to %s: %s' % (infile, outfile, e))
 137       # Probably a different file system.
 138       logging.warning(
 139           'Failed to hardlink, failing back to copy %s to %s' % (
 140             infile, outfile))
 141       readable_copy(outfile, infile)
 142
 143
 144 def _set_write_bit(path, read_only):
 145   """Sets or resets the executable bit on a file or directory."""
 146   mode = os.lstat(path).st_mode
 147   if read_only:
 148     mode = mode & 0500
 149   else:
 150     mode = mode | 0200
 151   if hasattr(os, 'lchmod'):
 152     os.lchmod(path, mode)  # pylint: disable=E1101
 153   else:
 154     if stat.S_ISLNK(mode):
 155       # Skip symlink without lchmod() support.
 156       logging.debug('Can\'t change +w bit on symlink %s' % path)
 157       return
 158
 159     # TODO(maruel): Implement proper DACL modification on Windows.
 160     os.chmod(path, mode)
 161
 162
 163 def make_writable(root, read_only):
 164   """Toggle the writable bit on a directory tree."""
 165   assert os.path.isabs(root), root
 166   for dirpath, dirnames, filenames in os.walk(root, topdown=True):
 167     for filename in filenames:
 168       _set_write_bit(os.path.join(dirpath, filename), read_only)
 169
 170     for dirname in dirnames:
 171       _set_write_bit(os.path.join(dirpath, dirname), read_only)
 172
 173
 174 def rmtree(root):
 175   """Wrapper around shutil.rmtree() to retry automatically on Windows."""
 176   make_writable(root, False)
 177   if sys.platform == 'win32':
 178     for i in range(3):
 179       try:
 180         shutil.rmtree(root)
 181         break
 182       except WindowsError:  # pylint: disable=E0602
 183         delay = (i+1)*2
 184         print >> sys.stderr, (
 185             'The test has subprocess outliving it. Sleep %d seconds.' % delay)
 186         time.sleep(delay)
 187   else:
 188     shutil.rmtree(root)
 189
 190
 191 def try_remove(filepath):
 192   """Removes a file without crashing even if it doesn't exist."""
 193   try:
 194     os.remove(filepath)
 195   except OSError:
 196     pass
 197
 198
 199 def is_same_filesystem(path1, path2):
 200   """Returns True if both paths are on the same filesystem.
 201
 202   This is required to enable the use of hardlinks.
 203   """
 204   assert os.path.isabs(path1), path1
 205   assert os.path.isabs(path2), path2
 206   if sys.platform == 'win32':
 207     # If the drive letter mismatches, assume it's a separate partition.
 208     # TODO(maruel): It should look at the underlying drive, a drive letter could
 209     # be a mount point to a directory on another drive.
 210     assert re.match(r'^[a-zA-Z]\:\\.*', path1), path1
 211     assert re.match(r'^[a-zA-Z]\:\\.*', path2), path2
 212     if path1[0].lower() != path2[0].lower():
 213       return False
 214   return os.stat(path1).st_dev == os.stat(path2).st_dev
 215
 216
 217 def get_free_space(path):
 218   """Returns the number of free bytes."""
 219   if sys.platform == 'win32':
 220     free_bytes = ctypes.c_ulonglong(0)
 221     ctypes.windll.kernel32.GetDiskFreeSpaceExW(
 222         ctypes.c_wchar_p(path), None, None, ctypes.pointer(free_bytes))
 223     return free_bytes.value
 224   # For OSes other than Windows.
 225   f = os.statvfs(path)  # pylint: disable=E1101
 226   return f.f_bfree * f.f_frsize
 227
 228
 229 def make_temp_dir(prefix, root_dir):
 230   """Returns a temporary directory on the same file system as root_dir."""
 231   base_temp_dir = None
 232   if not is_same_filesystem(root_dir, tempfile.gettempdir()):
 233     base_temp_dir = os.path.dirname(root_dir)
 234   return tempfile.mkdtemp(prefix=prefix, dir=base_temp_dir)
 235
 236
 237 class CachePolicies(object):
 238   def __init__(self, max_cache_size, min_free_space, max_items):
 239     """
 240     Arguments:
 241     - max_cache_size: Trim if the cache gets larger than this value. If 0, the
 242                       cache is effectively a leak.
 243     - min_free_space: Trim if disk free space becomes lower than this value. If
 244                       0, it unconditionally fill the disk.
 245     - max_items: Maximum number of items to keep in the cache. If 0, do not
 246                  enforce a limit.
 247     """
 248     self.max_cache_size = max_cache_size
 249     self.min_free_space = min_free_space
 250     self.max_items = max_items
 251
 252
 253 class DiskCache(isolateserver.LocalCache):
 254   """Stateful LRU cache in a flat hash table in a directory.
 255
 256   Saves its state as json file.
 257   """
 258   STATE_FILE = 'state.json'
 259
 260   def __init__(self, cache_dir, policies, algo):
 261     """
 262     Arguments:
 263       cache_dir: directory where to place the cache.
 264       policies: cache retention policies.
 265       algo: hashing algorithm used.
 266     """
 267     super(DiskCache, self).__init__()
 268     self.algo = algo
 269     self.cache_dir = cache_dir
 270     self.policies = policies
 271     self.state_file = os.path.join(cache_dir, self.STATE_FILE)
 272
 273     # All protected methods (starting with '_') except _path should be called
 274     # with this lock locked.
 275     self._lock = threading_utils.LockWithAssert()
 276     self._lru = lru.LRUDict()
 277
 278     # Profiling values.
 279     self._added = []
 280     self._removed = []
 281     self._free_disk = 0
 282
 283     with tools.Profiler('Setup'):
 284       with self._lock:
 285         self._load()
 286
 287   def __enter__(self):
 288     return self
 289
 290   def __exit__(self, _exc_type, _exec_value, _traceback):
 291     with tools.Profiler('CleanupTrimming'):
 292       with self._lock:
 293         self._trim()
 294
 295         logging.info(
 296             '%5d (%8dkb) added',
 297             len(self._added), sum(self._added) / 1024)
 298         logging.info(
 299             '%5d (%8dkb) current',
 300             len(self._lru),
 301             sum(self._lru.itervalues()) / 1024)
 302         logging.info(
 303             '%5d (%8dkb) removed',
 304             len(self._removed), sum(self._removed) / 1024)
 305         logging.info(
 306             '       %8dkb free',
 307             self._free_disk / 1024)
 308     return False
 309
 310   def cached_set(self):
 311     with self._lock:
 312       return self._lru.keys_set()
 313
 314   def touch(self, digest, size):
 315     # Verify an actual file is valid. Note that is doesn't compute the hash so
 316     # it could still be corrupted. Do it outside the lock.
 317     if not isolateserver.is_valid_file(self._path(digest), size):
 318       return False
 319
 320     # Update it's LRU position.
 321     with self._lock:
 322       if digest not in self._lru:
 323         return False
 324       self._lru.touch(digest)
 325     return True
 326
 327   def evict(self, digest):
 328     with self._lock:
 329       self._lru.pop(digest)
 330       self._delete_file(digest, isolateserver.UNKNOWN_FILE_SIZE)
 331
 332   def read(self, digest):
 333     with open(self._path(digest), 'rb') as f:
 334       return f.read()
 335
 336   def write(self, digest, content):
 337     path = self._path(digest)
 338     try:
 339       size = isolateserver.file_write(path, content)
 340     except:
 341       # There are two possible places were an exception can occur:
 342       #   1) Inside |content| generator in case of network or unzipping errors.
 343       #   2) Inside file_write itself in case of disk IO errors.
 344       # In any case delete an incomplete file and propagate the exception to
 345       # caller, it will be logged there.
 346       try_remove(path)
 347       raise
 348     with self._lock:
 349       self._add(digest, size)
 350
 351   def link(self, digest, dest, file_mode=None):
 352     link_file(dest, self._path(digest), HARDLINK)
 353     if file_mode is not None:
 354       os.chmod(dest, file_mode)
 355
 356   def _load(self):
 357     """Loads state of the cache from json file."""
 358     self._lock.assert_locked()
 359
 360     if not os.path.isdir(self.cache_dir):
 361       os.makedirs(self.cache_dir)
 362
 363     # Load state of the cache.
 364     if os.path.isfile(self.state_file):
 365       try:
 366         self._lru = lru.LRUDict.load(self.state_file)
 367       except ValueError as err:
 368         logging.error('Failed to load cache state: %s' % (err,))
 369         # Don't want to keep broken state file.
 370         os.remove(self.state_file)
 371
 372     # Ensure that all files listed in the state still exist and add new ones.
 373     previous = self._lru.keys_set()
 374     unknown = []
 375     for filename in os.listdir(self.cache_dir):
 376       if filename == self.STATE_FILE:
 377         continue
 378       if filename in previous:
 379         previous.remove(filename)
 380         continue
 381       # An untracked file.
 382       if not isolateserver.is_valid_hash(filename, self.algo):
 383         logging.warning('Removing unknown file %s from cache', filename)
 384         try_remove(self._path(filename))
 385         continue
 386       # File that's not referenced in 'state.json'.
 387       # TODO(vadimsh): Verify its SHA1 matches file name.
 388       logging.warning('Adding unknown file %s to cache', filename)
 389       unknown.append(filename)
 390
 391     if unknown:
 392       # Add as oldest files. They will be deleted eventually if not accessed.
 393       self._add_oldest_list(unknown)
 394       logging.warning('Added back %d unknown files', len(unknown))
 395
 396     if previous:
 397       # Filter out entries that were not found.
 398       logging.warning('Removed %d lost files', len(previous))
 399       for filename in previous:
 400         self._lru.pop(filename)
 401     self._trim()
 402
 403   def _save(self):
 404     """Saves the LRU ordering."""
 405     self._lock.assert_locked()
 406     self._lru.save(self.state_file)
 407
 408   def _trim(self):
 409     """Trims anything we don't know, make sure enough free space exists."""
 410     self._lock.assert_locked()
 411
 412     # Ensure maximum cache size.
 413     if self.policies.max_cache_size:
 414       total_size = sum(self._lru.itervalues())
 415       while total_size > self.policies.max_cache_size:
 416         total_size -= self._remove_lru_file()
 417
 418     # Ensure maximum number of items in the cache.
 419     if self.policies.max_items and len(self._lru) > self.policies.max_items:
 420       for _ in xrange(len(self._lru) - self.policies.max_items):
 421         self._remove_lru_file()
 422
 423     # Ensure enough free space.
 424     self._free_disk = get_free_space(self.cache_dir)
 425     trimmed_due_to_space = False
 426     while (
 427         self.policies.min_free_space and
 428         self._lru and
 429         self._free_disk < self.policies.min_free_space):
 430       trimmed_due_to_space = True
 431       self._remove_lru_file()
 432       self._free_disk = get_free_space(self.cache_dir)
 433     if trimmed_due_to_space:
 434       total = sum(self._lru.itervalues())
 435       logging.warning(
 436           'Trimmed due to not enough free disk space: %.1fkb free, %.1fkb '
 437           'cache (%.1f%% of its maximum capacity)',
 438           self._free_disk / 1024.,
 439           total / 1024.,
 440           100. * self.policies.max_cache_size / float(total),
 441           )
 442     self._save()
 443
 444   def _path(self, digest):
 445     """Returns the path to one item."""
 446     return os.path.join(self.cache_dir, digest)
 447
 448   def _remove_lru_file(self):
 449     """Removes the last recently used file and returns its size."""
 450     self._lock.assert_locked()
 451     digest, size = self._lru.pop_oldest()
 452     self._delete_file(digest, size)
 453     return size
 454
 455   def _add(self, digest, size=isolateserver.UNKNOWN_FILE_SIZE):
 456     """Adds an item into LRU cache marking it as a newest one."""
 457     self._lock.assert_locked()
 458     if size == isolateserver.UNKNOWN_FILE_SIZE:
 459       size = os.stat(self._path(digest)).st_size
 460     self._added.append(size)
 461     self._lru.add(digest, size)
 462
 463   def _add_oldest_list(self, digests):
 464     """Adds a bunch of items into LRU cache marking them as oldest ones."""
 465     self._lock.assert_locked()
 466     pairs = []
 467     for digest in digests:
 468       size = os.stat(self._path(digest)).st_size
 469       self._added.append(size)
 470       pairs.append((digest, size))
 471     self._lru.batch_insert_oldest(pairs)
 472
 473   def _delete_file(self, digest, size=isolateserver.UNKNOWN_FILE_SIZE):
 474     """Deletes cache file from the file system."""
 475     self._lock.assert_locked()
 476     try:
 477       if size == isolateserver.UNKNOWN_FILE_SIZE:
 478         size = os.stat(self._path(digest)).st_size
 479       os.remove(self._path(digest))
 480       self._removed.append(size)
 481     except OSError as e:
 482       logging.error('Error attempting to delete a file %s:\n%s' % (digest, e))
 483
 484
 485 def run_tha_test(isolated_hash, storage, cache, algo, outdir):
 486   """Downloads the dependencies in the cache, hardlinks them into a |outdir|
 487   and runs the executable.
 488   """
 489   try:
 490     try:
 491       settings = isolateserver.fetch_isolated(
 492           isolated_hash=isolated_hash,
 493           storage=storage,
 494           cache=cache,
 495           algo=algo,
 496           outdir=outdir,
 497           os_flavor=get_flavor(),
 498           require_command=True)
 499     except isolateserver.ConfigError as e:
 500       print >> sys.stderr, str(e)
 501       return 1
 502
 503     if settings.read_only:
 504       logging.info('Making files read only')
 505       make_writable(outdir, True)
 506     cwd = os.path.normpath(os.path.join(outdir, settings.relative_cwd))
 507     logging.info('Running %s, cwd=%s' % (settings.command, cwd))
 508
 509     # TODO(csharp): This should be specified somewhere else.
 510     # TODO(vadimsh): Pass it via 'env_vars' in manifest.
 511     # Add a rotating log file if one doesn't already exist.
 512     env = os.environ.copy()
 513     if MAIN_DIR:
 514       env.setdefault('RUN_TEST_CASES_LOG_FILE',
 515           os.path.join(MAIN_DIR, RUN_TEST_CASES_LOG))
 516     try:
 517       with tools.Profiler('RunTest'):
 518         return subprocess.call(settings.command, cwd=cwd, env=env)
 519     except OSError:
 520       print >> sys.stderr, 'Failed to run %s; cwd=%s' % (settings.command, cwd)
 521       return 1
 522   finally:
 523     if outdir:
 524       rmtree(outdir)
 525
 526
 527 def main():
 528   tools.disable_buffering()
 529   parser = tools.OptionParserWithLogging(
 530       usage='%prog <options>',
 531       version=__version__,
 532       log_file=RUN_ISOLATED_LOG_FILE)
 533
 534   group = optparse.OptionGroup(parser, 'Data source')
 535   group.add_option(
 536       '-s', '--isolated',
 537       metavar='FILE',
 538       help='File/url describing what to map or run')
 539   group.add_option(
 540       '-H', '--hash',
 541       help='Hash of the .isolated to grab from the hash table')
 542   group.add_option(
 543       '-I', '--isolate-server',
 544       metavar='URL', default='',
 545       help='Isolate server to use')
 546   group.add_option(
 547       '-n', '--namespace',
 548       default='default-gzip',
 549       help='namespace to use when using isolateserver, default: %default')
 550   parser.add_option_group(group)
 551
 552   group = optparse.OptionGroup(parser, 'Cache management')
 553   group.add_option(
 554       '--cache',
 555       default='cache',
 556       metavar='DIR',
 557       help='Cache directory, default=%default')
 558   group.add_option(
 559       '--max-cache-size',
 560       type='int',
 561       metavar='NNN',
 562       default=20*1024*1024*1024,
 563       help='Trim if the cache gets larger than this value, default=%default')
 564   group.add_option(
 565       '--min-free-space',
 566       type='int',
 567       metavar='NNN',
 568       default=2*1024*1024*1024,
 569       help='Trim if disk free space becomes lower than this value, '
 570            'default=%default')
 571   group.add_option(
 572       '--max-items',
 573       type='int',
 574       metavar='NNN',
 575       default=100000,
 576       help='Trim if more than this number of items are in the cache '
 577            'default=%default')
 578   parser.add_option_group(group)
 579
 580   options, args = parser.parse_args()
 581
 582   if bool(options.isolated) == bool(options.hash):
 583     logging.debug('One and only one of --isolated or --hash is required.')
 584     parser.error('One and only one of --isolated or --hash is required.')
 585   if args:
 586     logging.debug('Unsupported args %s' % ' '.join(args))
 587     parser.error('Unsupported args %s' % ' '.join(args))
 588   if not options.isolate_server:
 589     parser.error('--isolate-server is required.')
 590
 591   options.cache = os.path.abspath(options.cache)
 592   policies = CachePolicies(
 593       options.max_cache_size, options.min_free_space, options.max_items)
 594   storage = isolateserver.get_storage(options.isolate_server, options.namespace)
 595   algo = isolateserver.get_hash_algo(options.namespace)
 596
 597   try:
 598     # |options.cache| may not exist until DiskCache() instance is created.
 599     cache = DiskCache(options.cache, policies, algo)
 600     outdir = make_temp_dir('run_tha_test', options.cache)
 601     return run_tha_test(
 602         options.isolated or options.hash, storage, cache, algo, outdir)
 603   except Exception as e:
 604     # Make sure any exception is logged.
 605     logging.exception(e)
 606     return 1
 607
 608
 609 if __name__ == '__main__':
 610   # Ensure that we are always running with the correct encoding.
 611   fix_encoding.fix_encoding()
 612   sys.exit(main())