src/tools/swarming_client/tools/run_swarm_tests_on_swarm.py

   1 #!/usr/bin/env python
   2 # Copyright 2012 The Swarming Authors. All rights reserved.
   3 # Use of this source code is governed under the Apache License, Version 2.0 that
   4 # can be found in the LICENSE file.
   5
   6 """Runs the whole set unit tests on swarm.
   7
   8 This is done in a few steps:
   9   - Archive the whole directory as a single .isolated file.
  10   - Create one test-specific .isolated for each test to run. The file is created
  11     directly and archived manually with isolateserver.py.
  12   - Trigger each of these test-specific .isolated file per OS.
  13   - Get all results out of order.
  14 """
  15
  16 import datetime
  17 import glob
  18 import getpass
  19 import logging
  20 import optparse
  21 import os
  22 import shutil
  23 import subprocess
  24 import sys
  25 import tempfile
  26 import time
  27
  28 BASE_DIR = os.path.dirname(os.path.abspath(__file__))
  29 ROOT_DIR = os.path.dirname(BASE_DIR)
  30
  31 sys.path.insert(0, ROOT_DIR)
  32
  33 from utils import threading_utils
  34 from utils import tools
  35
  36
  37 def check_output(cmd, cwd):
  38   return subprocess.check_output([sys.executable] + cmd, cwd=cwd)
  39
  40
  41 def capture(cmd, cwd):
  42   start = time.time()
  43   p = subprocess.Popen([sys.executable] + cmd, cwd=cwd, stdout=subprocess.PIPE)
  44   out = p.communicate()[0]
  45   return p.returncode, out, time.time() - start
  46
  47
  48 def archive_tree(root, isolate_server):
  49   """Archives a whole tree and return the sha1 of the .isolated file.
  50
  51   Manually creates a temporary isolated file and archives it.
  52   """
  53   logging.info('archive_tree(%s)', root)
  54   cmd = [
  55       'isolateserver.py', 'archive', '--isolate-server', isolate_server, root,
  56   ]
  57   if logging.getLogger().isEnabledFor(logging.INFO):
  58     cmd.append('--verbose')
  59   out = check_output(cmd, root)
  60   return out.split()[0]
  61
  62
  63 def archive_isolated_triggers(cwd, isolate_server, tree_isolated, tests):
  64   """Creates and archives all the .isolated files for the tests at once.
  65
  66   Archiving them in one batch is faster than archiving each file individually.
  67   Also the .isolated files can be reused across OSes, reducing the amount of
  68   I/O.
  69
  70   Returns:
  71     list of (test, sha1) tuples.
  72   """
  73   logging.info(
  74       'archive_isolated_triggers(%s, %s, %s)', cwd, tree_isolated, tests)
  75   tempdir = tempfile.mkdtemp(prefix='run_swarm_tests_on_swarm_')
  76   try:
  77     isolateds = []
  78     for test in tests:
  79       test_name = os.path.basename(test)
  80       # Creates a manual .isolated file. See
  81       # https://code.google.com/p/swarming/wiki/IsolatedDesign for more details.
  82       isolated = {
  83         'algo': 'sha-1',
  84         'command': ['python', test],
  85         'includes': [tree_isolated],
  86         'version': '1.0',
  87       }
  88       v = os.path.join(tempdir, test_name + '.isolated')
  89       tools.write_json(v, isolated, True)
  90       isolateds.append(v)
  91     cmd = [
  92         'isolateserver.py', 'archive', '--isolate-server', isolate_server,
  93     ] + isolateds
  94     if logging.getLogger().isEnabledFor(logging.INFO):
  95       cmd.append('--verbose')
  96     items = [i.split() for i in check_output(cmd, cwd).splitlines()]
  97     assert len(items) == len(tests)
  98     assert all(
  99         items[i][1].endswith(os.path.basename(tests[i]) + '.isolated')
 100         for i in xrange(len(tests)))
 101     return zip(tests, [i[0] for i in items])
 102   finally:
 103     shutil.rmtree(tempdir)
 104
 105
 106 def trigger(
 107     cwd, swarm_server, isolate_server, task_name, platform, isolated_hash):
 108   """Triggers a specified .isolated file."""
 109   cmd = [
 110       'swarming.py',
 111       'trigger',
 112       '--swarming', swarm_server,
 113       '--isolate-server', isolate_server,
 114       '--dimension', 'os', platform,
 115       '--task-name', task_name,
 116       isolated_hash,
 117   ]
 118   return capture(cmd, cwd)
 119
 120
 121 def collect(cwd, swarm_server, task_name):
 122   cmd = [
 123       'swarming.py',
 124       'collect',
 125       '--swarming', swarm_server,
 126       task_name,
 127   ]
 128   return capture(cmd, cwd)
 129
 130
 131 class Runner(object):
 132   def __init__(self, isolate_server, swarm_server, add_task, progress):
 133     self.isolate_server = isolate_server
 134     self.swarm_server = swarm_server
 135     self.add_task = add_task
 136     self.progress = progress
 137     self.prefix = (
 138         getpass.getuser() + '-' + datetime.datetime.now().isoformat() + '-')
 139
 140   def trigger(self, task_name, platform, isolated_hash):
 141     returncode, stdout, duration = trigger(
 142         ROOT_DIR,
 143         self.swarm_server,
 144         self.isolate_server,
 145         task_name,
 146         platform,
 147         isolated_hash)
 148     step_name = '%s (%3.2fs)' % (task_name, duration)
 149     if returncode:
 150       line = 'Failed to trigger %s\n%s' % (step_name, stdout)
 151       self.progress.update_item(line, index=1)
 152       return
 153     self.progress.update_item('Triggered %s' % step_name, index=1)
 154     self.add_task(0, self.collect, task_name, platform)
 155
 156   def collect(self, task_name, platform):
 157     returncode, stdout, duration = collect(
 158         ROOT_DIR, self.swarm_server, task_name)
 159     step_name = '%s (%3.2fs)' % (task_name, duration)
 160     if returncode:
 161       # Only print the output for failures, successes are unexciting.
 162       self.progress.update_item(
 163           'Failed %s:\n%s' % (step_name, stdout), index=1)
 164       return (task_name, platform, stdout)
 165     self.progress.update_item('Passed %s' % step_name, index=1)
 166
 167
 168 def run_swarm_tests_on_swarm(swarm_server, isolate_server, oses, tests, logs):
 169   """Archives, triggers swarming jobs and gets results."""
 170   start = time.time()
 171   # First, archive the whole tree.
 172   tree_isolated = archive_tree(ROOT_DIR, isolate_server)
 173
 174   # Create and archive all the .isolated files.
 175   isolateds = archive_isolated_triggers(
 176       ROOT_DIR, isolate_server, tree_isolated, tests)
 177   logging.debug('%s', isolateds)
 178   print('Archival took %3.2fs' % (time.time() - start))
 179
 180   # Trigger all the jobs and get results. This is parallelized in worker
 181   # threads.
 182   runs = len(isolateds) * len(oses)
 183   # triger + collect
 184   total = 2 * runs
 185   columns = [('index', 0), ('size', total)]
 186   progress = threading_utils.Progress(columns)
 187   progress.use_cr_only = False
 188   failed_tests = []
 189   with threading_utils.ThreadPoolWithProgress(
 190       progress, runs, runs, total) as pool:
 191     start = time.time()
 192     runner = Runner(isolate_server, swarm_server, pool.add_task, progress)
 193     for test_path, isolated in isolateds:
 194       test_name = os.path.basename(test_path).split('.')[0]
 195       for platform in oses:
 196         task_name = '%s/%s/%s' % (test_name, platform, isolated)
 197         pool.add_task(0, runner.trigger, task_name, platform, isolated)
 198
 199     for failed_test in pool.iter_results():
 200       # collect() only return test case failures.
 201       test_name, platform, stdout = failed_test
 202       failed_tests.append(test_name)
 203       if logs:
 204         # Write the logs are they are retrieved.
 205         if not os.path.isdir(logs):
 206           os.makedirs(logs)
 207         name = '%s_%s.log' % (platform, test_name.split('/', 1)[0])
 208         with open(os.path.join(logs, name), 'wb') as f:
 209           f.write(stdout)
 210   duration = time.time() - start
 211   print('\nCompleted in %3.2fs' % duration)
 212   if failed_tests:
 213     print('Detected the following failures:')
 214     for test in sorted(failed_tests):
 215       print('  %s' % test)
 216   return bool(failed_tests)
 217
 218
 219 def main():
 220   parser = optparse.OptionParser(description=sys.modules[__name__].__doc__)
 221   parser.add_option(
 222       '-I', '--isolate-server',
 223       metavar='URL', default='',
 224       help='Isolate server to use')
 225   parser.add_option(
 226       '-S', '--swarming',
 227       metavar='URL', default='',
 228       help='Swarming server to use')
 229   parser.add_option(
 230       '-l', '--logs',
 231       help='Destination where to store the failure logs (recommended)')
 232   parser.add_option('-o', '--os', help='Run tests only on this OS')
 233   parser.add_option(
 234       '-t', '--test', action='append',
 235       help='Run only these test, can be specified multiple times')
 236   parser.add_option('-v', '--verbose', action='store_true')
 237   options, args = parser.parse_args()
 238   if args:
 239     parser.error('Unsupported argument %s' % args)
 240   if options.verbose:
 241     os.environ['ISOLATE_DEBUG'] = '1'
 242
 243   if not options.isolate_server:
 244     parser.error('--isolate-server is required.')
 245   if not options.swarming:
 246     parser.error('--swarming is required.')
 247
 248   logging.basicConfig(level=logging.DEBUG if options.verbose else logging.ERROR)
 249
 250   oses = ['Linux', 'Mac', 'Windows']
 251   tests = [
 252       os.path.relpath(i, ROOT_DIR)
 253       for i in (
 254       glob.glob(os.path.join(ROOT_DIR, 'tests', '*_test.py')) +
 255       glob.glob(os.path.join(ROOT_DIR, 'googletest', 'tests', '*_test.py')))
 256   ]
 257   valid_tests = sorted(map(os.path.basename, tests))
 258   assert len(valid_tests) == len(set(valid_tests)), (
 259       'Can\'t have 2 tests with the same base name')
 260
 261   if options.test:
 262     for t in options.test:
 263       if not t in valid_tests:
 264         parser.error(
 265             '--test %s is unknown. Valid values are:\n%s' % (
 266               t, '\n'.join('  ' + i for i in valid_tests)))
 267     filters = tuple(os.path.sep + t for t in options.test)
 268     tests = [t for t in tests if t.endswith(filters)]
 269
 270   if options.os:
 271     if options.os not in oses:
 272       parser.error(
 273           '--os %s is unknown. Valid values are %s' % (
 274             options.os, ', '.join(sorted(oses))))
 275     oses = [options.os]
 276
 277   if sys.platform in ('win32', 'cygwin'):
 278     # If we are on Windows, don't generate the tests for Linux and Mac since
 279     # they use symlinks and we can't create symlinks on windows.
 280     oses = ['Windows']
 281     if options.os != 'win32':
 282       print('Linux and Mac tests skipped since running on Windows.')
 283
 284   return run_swarm_tests_on_swarm(
 285       options.swarming,
 286       options.isolate_server,
 287       oses,
 288       tests,
 289       options.logs)
 290
 291
 292 if __name__ == '__main__':
 293   sys.exit(main())