src/tools/swarming_client/swarming.py

   1 #!/usr/bin/env python
   2 # Copyright 2013 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 """Client tool to trigger tasks or retrieve results from a Swarming server."""
   7
   8 __version__ = '0.4.4'
   9
  10 import datetime
  11 import getpass
  12 import hashlib
  13 import json
  14 import logging
  15 import os
  16 import shutil
  17 import subprocess
  18 import sys
  19 import time
  20 import urllib
  21
  22 from third_party import colorama
  23 from third_party.depot_tools import fix_encoding
  24 from third_party.depot_tools import subcommand
  25
  26 from utils import file_path
  27 from third_party.chromium import natsort
  28 from utils import net
  29 from utils import threading_utils
  30 from utils import tools
  31 from utils import zip_package
  32
  33 import auth
  34 import isolateserver
  35 import run_isolated
  36
  37
  38 ROOT_DIR = os.path.dirname(os.path.abspath(__file__))
  39 TOOLS_PATH = os.path.join(ROOT_DIR, 'tools')
  40
  41
  42 # The default time to wait for a shard to finish running.
  43 DEFAULT_SHARD_WAIT_TIME = 80 * 60.
  44
  45
  46 NO_OUTPUT_FOUND = (
  47   'No output produced by the task, it may have failed to run.\n'
  48   '\n')
  49
  50
  51 class Failure(Exception):
  52   """Generic failure."""
  53   pass
  54
  55
  56 class Manifest(object):
  57   """Represents a Swarming task manifest.
  58
  59   Also includes code to zip code and upload itself.
  60   """
  61   def __init__(
  62       self, isolate_server, namespace, isolated_hash, task_name, shards, env,
  63       dimensions, working_dir, deadline, verbose, profile, priority):
  64     """Populates a manifest object.
  65       Args:
  66         isolate_server - isolate server url.
  67         namespace - isolate server namespace to use.
  68         isolated_hash - The manifest's sha-1 that the slave is going to fetch.
  69         task_name - The name to give the task request.
  70         shards - The number of swarming shards to request.
  71         env - environment variables to set.
  72         dimensions - dimensions to filter the task on.
  73         working_dir - Relative working directory to start the script.
  74         deadline - maximum pending time before this task expires.
  75         verbose - if True, have the slave print more details.
  76         profile - if True, have the slave print more timing data.
  77         priority - int between 0 and 1000, lower the higher priority.
  78     """
  79     self.isolate_server = isolate_server
  80     self.namespace = namespace
  81     # The reason is that swarm_bot doesn't understand compressed data yet. So
  82     # the data to be downloaded by swarm_bot is in 'default', independent of
  83     # what run_isolated.py is going to fetch.
  84     self.storage = isolateserver.get_storage(isolate_server, 'default')
  85
  86     self.isolated_hash = isolated_hash
  87     self.bundle = zip_package.ZipPackage(ROOT_DIR)
  88
  89     self._task_name = task_name
  90     self._shards = shards
  91     self._env = env.copy()
  92     self._dimensions = dimensions.copy()
  93     self._working_dir = working_dir
  94     self._deadline = deadline
  95
  96     self.verbose = bool(verbose)
  97     self.profile = bool(profile)
  98     self.priority = priority
  99
 100     self._isolate_item = None
 101     self._tasks = []
 102
 103   def add_task(self, task_name, actions, time_out=600):
 104     """Appends a new task as a TestObject to the swarming manifest file.
 105
 106     Tasks cannot be added once the manifest was uploaded.
 107
 108     See TestObject in services/swarming/src/common/test_request_message.py for
 109     the valid format.
 110     """
 111     assert not self._isolate_item
 112     self._tasks.append(
 113         {
 114           'action': actions,
 115           'decorate_output': self.verbose,
 116           'test_name': task_name,
 117           'time_out': time_out,
 118         })
 119
 120   def to_json(self):
 121     """Exports the current configuration into a swarm-readable manifest file.
 122
 123     The actual serialization format is defined as a TestCase object as described
 124     in services/swarming/src/common/test_request_message.py
 125
 126     This function doesn't mutate the object.
 127     """
 128     request = {
 129       'cleanup': 'root',
 130       'configurations': [
 131         # Is a TestConfiguration.
 132         {
 133           'config_name': 'isolated',
 134           'deadline_to_run': self._deadline,
 135           'dimensions': self._dimensions,
 136           'min_instances': self._shards,
 137           'priority': self.priority,
 138         },
 139       ],
 140       'data': [],
 141       'encoding': 'UTF-8',
 142       'env_vars': self._env,
 143       'restart_on_failure': True,
 144       'test_case_name': self._task_name,
 145       'tests': self._tasks,
 146       'working_dir': self._working_dir,
 147     }
 148     if self._isolate_item:
 149       request['data'].append(
 150           [
 151             self.storage.get_fetch_url(self._isolate_item),
 152             'swarm_data.zip',
 153           ])
 154     return json.dumps(request, sort_keys=True, separators=(',',':'))
 155
 156   @property
 157   def isolate_item(self):
 158     """Calling this property 'closes' the manifest and it can't be modified
 159     afterward.
 160     """
 161     if self._isolate_item is None:
 162       self._isolate_item = isolateserver.BufferItem(
 163           self.bundle.zip_into_buffer(), high_priority=True)
 164     return self._isolate_item
 165
 166
 167 def zip_and_upload(manifest):
 168   """Zips up all the files necessary to run a manifest and uploads to Swarming
 169   master.
 170   """
 171   try:
 172     start_time = time.time()
 173     with manifest.storage:
 174       uploaded = manifest.storage.upload_items([manifest.isolate_item])
 175     elapsed = time.time() - start_time
 176   except (IOError, OSError) as exc:
 177     tools.report_error('Failed to upload the zip file: %s' % exc)
 178     return False
 179
 180   if manifest.isolate_item in uploaded:
 181     logging.info('Upload complete, time elapsed: %f', elapsed)
 182   else:
 183     logging.info('Zip file already on server, time elapsed: %f', elapsed)
 184   return True
 185
 186
 187 def now():
 188   """Exists so it can be mocked easily."""
 189   return time.time()
 190
 191
 192 def get_task_keys(swarm_base_url, task_name):
 193   """Returns the Swarming task key for each shards of task_name."""
 194   key_data = urllib.urlencode([('name', task_name)])
 195   url = '%s/get_matching_test_cases?%s' % (swarm_base_url, key_data)
 196
 197   for _ in net.retry_loop(max_attempts=net.URL_OPEN_MAX_ATTEMPTS):
 198     result = net.url_read(url, retry_404=True)
 199     if result is None:
 200       raise Failure(
 201           'Error: Unable to find any task with the name, %s, on swarming server'
 202           % task_name)
 203
 204     # TODO(maruel): Compare exact string.
 205     if 'No matching' in result:
 206       logging.warning('Unable to find any task with the name, %s, on swarming '
 207                       'server' % task_name)
 208       continue
 209     return json.loads(result)
 210
 211   raise Failure(
 212       'Error: Unable to find any task with the name, %s, on swarming server'
 213       % task_name)
 214
 215
 216 def retrieve_results(base_url, task_key, timeout, should_stop):
 217   """Retrieves results for a single task_key."""
 218   assert isinstance(timeout, float), timeout
 219   params = [('r', task_key)]
 220   result_url = '%s/get_result?%s' % (base_url, urllib.urlencode(params))
 221   start = now()
 222   while True:
 223     if timeout and (now() - start) >= timeout:
 224       logging.error('retrieve_results(%s) timed out', base_url)
 225       return {}
 226     # Do retries ourselves.
 227     response = net.url_read(result_url, retry_404=False, retry_50x=False)
 228     if response is None:
 229       # Aggressively poll for results. Do not use retry_404 so
 230       # should_stop is polled more often.
 231       remaining = min(5, timeout - (now() - start)) if timeout else 5
 232       if remaining > 0:
 233         if should_stop.get():
 234           return {}
 235         net.sleep_before_retry(1, remaining)
 236     else:
 237       try:
 238         data = json.loads(response) or {}
 239       except (ValueError, TypeError):
 240         logging.warning(
 241             'Received corrupted data for task_key %s. Retrying.', task_key)
 242       else:
 243         if data['output']:
 244           return data
 245     if should_stop.get():
 246       return {}
 247
 248
 249 def yield_results(swarm_base_url, task_keys, timeout, max_threads):
 250   """Yields swarming task results from the swarming server as (index, result).
 251
 252   Duplicate shards are ignored, the first one to complete is returned.
 253
 254   max_threads is optional and is used to limit the number of parallel fetches
 255   done. Since in general the number of task_keys is in the range <=10, it's not
 256   worth normally to limit the number threads. Mostly used for testing purposes.
 257
 258   Yields:
 259     (index, result). In particular, 'result' is defined as the
 260     GetRunnerResults() function in services/swarming/server/test_runner.py.
 261   """
 262   shards_remaining = range(len(task_keys))
 263   number_threads = (
 264       min(max_threads, len(task_keys)) if max_threads else len(task_keys))
 265   should_stop = threading_utils.Bit()
 266   results_remaining = len(task_keys)
 267   with threading_utils.ThreadPool(number_threads, number_threads, 0) as pool:
 268     try:
 269       for task_key in task_keys:
 270         pool.add_task(
 271             0, retrieve_results, swarm_base_url, task_key, timeout, should_stop)
 272       while shards_remaining and results_remaining:
 273         result = pool.get_one_result()
 274         results_remaining -= 1
 275         if not result:
 276           # Failed to retrieve one key.
 277           logging.error('Failed to retrieve the results for a swarming key')
 278           continue
 279         shard_index = result['config_instance_index']
 280         if shard_index in shards_remaining:
 281           shards_remaining.remove(shard_index)
 282           yield shard_index, result
 283         else:
 284           logging.warning('Ignoring duplicate shard index %d', shard_index)
 285           # Pop the last entry, there's no such shard.
 286           shards_remaining.pop()
 287     finally:
 288       # Done, kill the remaining threads.
 289       should_stop.set()
 290
 291
 292 def chromium_setup(manifest):
 293   """Sets up the commands to run.
 294
 295   Highly chromium specific.
 296   """
 297   # Add uncompressed zip here. It'll be compressed as part of the package sent
 298   # to Swarming server.
 299   run_test_name = 'run_isolated.zip'
 300   manifest.bundle.add_buffer(run_test_name,
 301     run_isolated.get_as_zip_package().zip_into_buffer(compress=False))
 302
 303   cleanup_script_name = 'swarm_cleanup.py'
 304   manifest.bundle.add_file(os.path.join(TOOLS_PATH, cleanup_script_name),
 305     cleanup_script_name)
 306
 307   run_cmd = [
 308     'python', run_test_name,
 309     '--hash', manifest.isolated_hash,
 310     '--namespace', manifest.namespace,
 311   ]
 312   if file_path.is_url(manifest.isolate_server):
 313     run_cmd.extend(('--isolate-server', manifest.isolate_server))
 314   else:
 315     run_cmd.extend(('--indir', manifest.isolate_server))
 316
 317   if manifest.verbose or manifest.profile:
 318     # Have it print the profiling section.
 319     run_cmd.append('--verbose')
 320   manifest.add_task('Run Test', run_cmd)
 321
 322   # Clean up
 323   manifest.add_task('Clean Up', ['python', cleanup_script_name])
 324
 325
 326 def googletest_setup(env, shards):
 327   """Sets googletest specific environment variables."""
 328   if shards > 1:
 329     env = env.copy()
 330     env['GTEST_SHARD_INDEX'] = '%(instance_index)s'
 331     env['GTEST_TOTAL_SHARDS'] = '%(num_instances)s'
 332   return env
 333
 334
 335 def archive(isolate_server, namespace, isolated, algo, verbose):
 336   """Archives a .isolated and all the dependencies on the CAC."""
 337   logging.info('archive(%s, %s, %s)', isolate_server, namespace, isolated)
 338   tempdir = None
 339   if file_path.is_url(isolate_server):
 340     command = 'archive'
 341     flag = '--isolate-server'
 342   else:
 343     command = 'hashtable'
 344     flag = '--outdir'
 345
 346   print('Archiving: %s' % isolated)
 347   try:
 348     cmd = [
 349       sys.executable,
 350       os.path.join(ROOT_DIR, 'isolate.py'),
 351       command,
 352       flag, isolate_server,
 353       '--namespace', namespace,
 354       '--isolated', isolated,
 355     ]
 356     cmd.extend(['--verbose'] * verbose)
 357     logging.info(' '.join(cmd))
 358     if subprocess.call(cmd, verbose):
 359       return
 360     return isolateserver.hash_file(isolated, algo)
 361   finally:
 362     if tempdir:
 363       shutil.rmtree(tempdir)
 364
 365
 366 def process_manifest(
 367     swarming, isolate_server, namespace, isolated_hash, task_name, shards,
 368     dimensions, env, working_dir, deadline, verbose, profile, priority):
 369   """Processes the manifest file and send off the swarming task request."""
 370   try:
 371     manifest = Manifest(
 372         isolate_server=isolate_server,
 373         namespace=namespace,
 374         isolated_hash=isolated_hash,
 375         task_name=task_name,
 376         shards=shards,
 377         dimensions=dimensions,
 378         env=env,
 379         working_dir=working_dir,
 380         deadline=deadline,
 381         verbose=verbose,
 382         profile=profile,
 383         priority=priority)
 384   except ValueError as e:
 385     tools.report_error('Unable to process %s: %s' % (task_name, e))
 386     return 1
 387
 388   chromium_setup(manifest)
 389
 390   logging.info('Zipping up files...')
 391   if not zip_and_upload(manifest):
 392     return 1
 393
 394   logging.info('Server: %s', swarming)
 395   logging.info('Task name: %s', task_name)
 396   trigger_url = swarming + '/test'
 397   manifest_text = manifest.to_json()
 398   result = net.url_read(trigger_url, data={'request': manifest_text})
 399   if not result:
 400     tools.report_error(
 401         'Failed to trigger task %s\n%s' % (task_name, trigger_url))
 402     return 1
 403   try:
 404     json.loads(result)
 405   except (ValueError, TypeError) as e:
 406     msg = '\n'.join((
 407         'Failed to trigger task %s' % task_name,
 408         'Manifest: %s' % manifest_text,
 409         'Bad response: %s' % result,
 410         str(e)))
 411     tools.report_error(msg)
 412     return 1
 413   return 0
 414
 415
 416 def isolated_to_hash(isolate_server, namespace, arg, algo, verbose):
 417   """Archives a .isolated file if needed.
 418
 419   Returns the file hash to trigger and a bool specifying if it was a file (True)
 420   or a hash (False).
 421   """
 422   if arg.endswith('.isolated'):
 423     file_hash = archive(isolate_server, namespace, arg, algo, verbose)
 424     if not file_hash:
 425       tools.report_error('Archival failure %s' % arg)
 426       return None, True
 427     return file_hash, True
 428   elif isolateserver.is_valid_hash(arg, algo):
 429     return arg, False
 430   else:
 431     tools.report_error('Invalid hash %s' % arg)
 432     return None, False
 433
 434
 435 def trigger(
 436     swarming,
 437     isolate_server,
 438     namespace,
 439     file_hash_or_isolated,
 440     task_name,
 441     shards,
 442     dimensions,
 443     env,
 444     working_dir,
 445     deadline,
 446     verbose,
 447     profile,
 448     priority):
 449   """Sends off the hash swarming task requests."""
 450   file_hash, is_file = isolated_to_hash(
 451       isolate_server, namespace, file_hash_or_isolated, hashlib.sha1, verbose)
 452   if not file_hash:
 453     return 1, ''
 454   if not task_name:
 455     # If a file name was passed, use its base name of the isolated hash.
 456     # Otherwise, use user name as an approximation of a task name.
 457     if is_file:
 458       key = os.path.splitext(os.path.basename(file_hash_or_isolated))[0]
 459     else:
 460       key = getpass.getuser()
 461     task_name = '%s/%s/%s' % (
 462         key,
 463         '_'.join('%s=%s' % (k, v) for k, v in sorted(dimensions.iteritems())),
 464         file_hash)
 465
 466   env = googletest_setup(env, shards)
 467   # TODO(maruel): It should first create a request manifest object, then pass
 468   # it to a function to zip, archive and trigger.
 469   result = process_manifest(
 470       swarming=swarming,
 471       isolate_server=isolate_server,
 472       namespace=namespace,
 473       isolated_hash=file_hash,
 474       task_name=task_name,
 475       shards=shards,
 476       dimensions=dimensions,
 477       deadline=deadline,
 478       env=env,
 479       working_dir=working_dir,
 480       verbose=verbose,
 481       profile=profile,
 482       priority=priority)
 483   return result, task_name
 484
 485
 486 def decorate_shard_output(result, shard_exit_code):
 487   """Returns wrapped output for swarming task shard."""
 488   tag = 'index %s (machine tag: %s, id: %s)' % (
 489       result['config_instance_index'],
 490       result['machine_id'],
 491       result.get('machine_tag', 'unknown'))
 492   return (
 493     '\n'
 494     '================================================================\n'
 495     'Begin output from shard %s\n'
 496     '================================================================\n'
 497     '\n'
 498     '%s'
 499     '================================================================\n'
 500     'End output from shard %s. Return %d\n'
 501     '================================================================\n'
 502     ) % (tag, result['output'] or NO_OUTPUT_FOUND, tag, shard_exit_code)
 503
 504
 505 def collect(url, task_name, timeout, decorate):
 506   """Retrieves results of a Swarming task."""
 507   logging.info('Collecting %s', task_name)
 508   task_keys = get_task_keys(url, task_name)
 509   if not task_keys:
 510     raise Failure('No task keys to get results with.')
 511
 512   exit_code = None
 513   for _index, output in yield_results(url, task_keys, timeout, None):
 514     shard_exit_codes = (output['exit_codes'] or '1').split(',')
 515     shard_exit_code = max(int(i) for i in shard_exit_codes)
 516     if decorate:
 517       print decorate_shard_output(output, shard_exit_code)
 518     else:
 519       print(
 520           '%s/%s: %s' % (
 521               output['machine_id'],
 522               output['machine_tag'],
 523               output['exit_codes']))
 524       print(''.join('  %s\n' % l for l in output['output'].splitlines()))
 525     exit_code = exit_code or shard_exit_code
 526   return exit_code if exit_code is not None else 1
 527
 528
 529 def add_filter_options(parser):
 530   parser.filter_group = tools.optparse.OptionGroup(parser, 'Filtering slaves')
 531   parser.filter_group.add_option(
 532       '-d', '--dimension', default=[], action='append', nargs=2,
 533       dest='dimensions', metavar='FOO bar',
 534       help='dimension to filter on')
 535   parser.add_option_group(parser.filter_group)
 536
 537
 538 def add_trigger_options(parser):
 539   """Adds all options to trigger a task on Swarming."""
 540   isolateserver.add_isolate_server_options(parser, True)
 541   add_filter_options(parser)
 542
 543   parser.task_group = tools.optparse.OptionGroup(parser, 'Task properties')
 544   parser.task_group.add_option(
 545       '-w', '--working-dir', default='swarm_tests',
 546       help='Working directory on the swarming slave side. default: %default.')
 547   parser.task_group.add_option(
 548       '--working_dir', help=tools.optparse.SUPPRESS_HELP)
 549   parser.task_group.add_option(
 550       '-e', '--env', default=[], action='append', nargs=2, metavar='FOO bar',
 551       help='environment variables to set')
 552   parser.task_group.add_option(
 553       '--priority', type='int', default=100,
 554       help='The lower value, the more important the task is')
 555   parser.task_group.add_option(
 556       '--shards', type='int', default=1, help='number of shards to use')
 557   parser.task_group.add_option(
 558       '-T', '--task-name',
 559       help='Display name of the task. It uniquely identifies the task. '
 560            'Defaults to <base_name>/<dimensions>/<isolated hash> if an '
 561            'isolated file is provided, if a hash is provided, it defaults to '
 562            '<user>/<dimensions>/<isolated hash>')
 563   parser.task_group.add_option(
 564       '--deadline', type='int', default=6*60*60,
 565       help='Seconds to allow the task to be pending for a bot to run before '
 566            'this task request expires.')
 567   parser.add_option_group(parser.task_group)
 568   # TODO(maruel): This is currently written in a chromium-specific way.
 569   parser.group_logging.add_option(
 570       '--profile', action='store_true',
 571       default=bool(os.environ.get('ISOLATE_DEBUG')),
 572       help='Have run_isolated.py print profiling info')
 573
 574
 575 def process_trigger_options(parser, options, args):
 576   isolateserver.process_isolate_server_options(parser, options)
 577   if len(args) != 1:
 578     parser.error('Must pass one .isolated file or its hash (sha1).')
 579   options.dimensions = dict(options.dimensions)
 580   if not options.dimensions:
 581     parser.error('Please at least specify one --dimension')
 582
 583
 584 def add_collect_options(parser):
 585   parser.server_group.add_option(
 586       '-t', '--timeout',
 587       type='float',
 588       default=DEFAULT_SHARD_WAIT_TIME,
 589       help='Timeout to wait for result, set to 0 for no timeout; default: '
 590            '%default s')
 591   parser.group_logging.add_option(
 592       '--decorate', action='store_true', help='Decorate output')
 593
 594
 595 @subcommand.usage('task_name')
 596 def CMDcollect(parser, args):
 597   """Retrieves results of a Swarming task.
 598
 599   The result can be in multiple part if the execution was sharded. It can
 600   potentially have retries.
 601   """
 602   add_collect_options(parser)
 603   (options, args) = parser.parse_args(args)
 604   if not args:
 605     parser.error('Must specify one task name.')
 606   elif len(args) > 1:
 607     parser.error('Must specify only one task name.')
 608
 609   try:
 610     return collect(options.swarming, args[0], options.timeout, options.decorate)
 611   except Failure as e:
 612     tools.report_error(e)
 613     return 1
 614
 615
 616 def CMDquery(parser, args):
 617   """Returns information about the bots connected to the Swarming server."""
 618   add_filter_options(parser)
 619   parser.filter_group.add_option(
 620       '--dead-only', action='store_true',
 621       help='Only print dead bots, useful to reap them and reimage broken bots')
 622   parser.filter_group.add_option(
 623       '-k', '--keep-dead', action='store_true',
 624       help='Do not filter out dead bots')
 625   parser.filter_group.add_option(
 626       '-b', '--bare', action='store_true',
 627       help='Do not print out dimensions')
 628   options, args = parser.parse_args(args)
 629
 630   if options.keep_dead and options.dead_only:
 631     parser.error('Use only one of --keep-dead and --dead-only')
 632   service = net.get_http_service(options.swarming)
 633   data = service.json_request('GET', '/swarming/api/v1/bots')
 634   if data is None:
 635     print >> sys.stderr, 'Failed to access %s' % options.swarming
 636     return 1
 637   timeout = datetime.timedelta(seconds=data['machine_death_timeout'])
 638   utcnow = datetime.datetime.utcnow()
 639   for machine in natsort.natsorted(data['machines'], key=lambda x: x['tag']):
 640     last_seen = datetime.datetime.strptime(
 641         machine['last_seen'], '%Y-%m-%d %H:%M:%S')
 642     is_dead = utcnow - last_seen > timeout
 643     if options.dead_only:
 644       if not is_dead:
 645         continue
 646     elif not options.keep_dead and is_dead:
 647       continue
 648
 649     # If the user requested to filter on dimensions, ensure the bot has all the
 650     # dimensions requested.
 651     dimensions = machine['dimensions']
 652     for key, value in options.dimensions:
 653       if key not in dimensions:
 654         break
 655       # A bot can have multiple value for a key, for example,
 656       # {'os': ['Windows', 'Windows-6.1']}, so that --dimension os=Windows will
 657       # be accepted.
 658       if isinstance(dimensions[key], list):
 659         if value not in dimensions[key]:
 660           break
 661       else:
 662         if value != dimensions[key]:
 663           break
 664     else:
 665       print machine['tag']
 666       if not options.bare:
 667         print '  %s' % dimensions
 668   return 0
 669
 670
 671 @subcommand.usage('[hash|isolated]')
 672 def CMDrun(parser, args):
 673   """Triggers a task and wait for the results.
 674
 675   Basically, does everything to run a command remotely.
 676   """
 677   add_trigger_options(parser)
 678   add_collect_options(parser)
 679   options, args = parser.parse_args(args)
 680   process_trigger_options(parser, options, args)
 681
 682   try:
 683     result, task_name = trigger(
 684         swarming=options.swarming,
 685         isolate_server=options.isolate_server or options.indir,
 686         namespace=options.namespace,
 687         file_hash_or_isolated=args[0],
 688         task_name=options.task_name,
 689         shards=options.shards,
 690         dimensions=options.dimensions,
 691         env=dict(options.env),
 692         working_dir=options.working_dir,
 693         deadline=options.deadline,
 694         verbose=options.verbose,
 695         profile=options.profile,
 696         priority=options.priority)
 697   except Failure as e:
 698     tools.report_error(
 699         'Failed to trigger %s(%s): %s' %
 700         (options.task_name, args[0], e.args[0]))
 701     return 1
 702   if result:
 703     tools.report_error('Failed to trigger the task.')
 704     return result
 705   if task_name != options.task_name:
 706     print('Triggered task: %s' % task_name)
 707   try:
 708     return collect(
 709         options.swarming,
 710         task_name,
 711         options.timeout,
 712         options.decorate)
 713   except Failure as e:
 714     tools.report_error(e)
 715     return 1
 716
 717
 718 @subcommand.usage("(hash|isolated)")
 719 def CMDtrigger(parser, args):
 720   """Triggers a Swarming task.
 721
 722   Accepts either the hash (sha1) of a .isolated file already uploaded or the
 723   path to an .isolated file to archive, packages it if needed and sends a
 724   Swarming manifest file to the Swarming server.
 725
 726   If an .isolated file is specified instead of an hash, it is first archived.
 727   """
 728   add_trigger_options(parser)
 729   options, args = parser.parse_args(args)
 730   process_trigger_options(parser, options, args)
 731
 732   try:
 733     result, task_name = trigger(
 734         swarming=options.swarming,
 735         isolate_server=options.isolate_server or options.indir,
 736         namespace=options.namespace,
 737         file_hash_or_isolated=args[0],
 738         task_name=options.task_name,
 739         dimensions=options.dimensions,
 740         shards=options.shards,
 741         env=dict(options.env),
 742         working_dir=options.working_dir,
 743         deadline=options.deadline,
 744         verbose=options.verbose,
 745         profile=options.profile,
 746         priority=options.priority)
 747     if task_name != options.task_name and not result:
 748       print('Triggered task: %s' % task_name)
 749     return result
 750   except Failure as e:
 751     tools.report_error(e)
 752     return 1
 753
 754
 755 class OptionParserSwarming(tools.OptionParserWithLogging):
 756   def __init__(self, **kwargs):
 757     tools.OptionParserWithLogging.__init__(
 758         self, prog='swarming.py', **kwargs)
 759     self.server_group = tools.optparse.OptionGroup(self, 'Server')
 760     self.server_group.add_option(
 761         '-S', '--swarming',
 762         metavar='URL', default=os.environ.get('SWARMING_SERVER', ''),
 763         help='Swarming server to use')
 764     self.add_option_group(self.server_group)
 765     auth.add_auth_options(self)
 766
 767   def parse_args(self, *args, **kwargs):
 768     options, args = tools.OptionParserWithLogging.parse_args(
 769         self, *args, **kwargs)
 770     options.swarming = options.swarming.rstrip('/')
 771     if not options.swarming:
 772       self.error('--swarming is required.')
 773     auth.process_auth_options(self, options)
 774     return options, args
 775
 776
 777 def main(args):
 778   dispatcher = subcommand.CommandDispatcher(__name__)
 779   try:
 780     return dispatcher.execute(OptionParserSwarming(version=__version__), args)
 781   except Exception as e:
 782     tools.report_error(e)
 783     return 1
 784
 785
 786 if __name__ == '__main__':
 787   fix_encoding.fix_encoding()
 788   tools.disable_buffering()
 789   colorama.init()
 790   sys.exit(main(sys.argv[1:]))