src/tools/swarming_client/tools/swarming_load_test_client.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 """Triggers a ton of fake jobs to test its handling under high load.
   7
   8 Generates an histogram with the latencies to process the tasks and number of
   9 retries.
  10 """
  11
  12 import json
  13 import logging
  14 import optparse
  15 import os
  16 import random
  17 import string
  18 import sys
  19 import time
  20
  21 ROOT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
  22
  23 sys.path.insert(0, ROOT_DIR)
  24
  25 from third_party import colorama
  26
  27 import swarming
  28
  29 from utils import graph
  30 from utils import net
  31 from utils import threading_utils
  32
  33 import swarming_load_test_bot
  34
  35
  36 def print_results(results, columns, buckets):
  37   delays = [i for i in results if isinstance(i, float)]
  38   failures = [i for i in results if not isinstance(i, float)]
  39
  40   graph.print_histogram(
  41       graph.generate_histogram(delays, buckets), columns, '%5.3f')
  42   print('')
  43   print('Total items : %d' % len(results))
  44   average = 0
  45   if delays:
  46     average = sum(delays)/ len(delays)
  47   print('Average delay: %s' % graph.to_units(average))
  48   #print('Average overhead: %s' % graph.to_units(total_size / len(sizes)))
  49   print('')
  50   if failures:
  51     print('')
  52     print('%sFAILURES%s:' % (colorama.Fore.RED, colorama.Fore.RESET))
  53     print('\n'.join('  %s' % i for i in failures))
  54
  55
  56 def trigger_task(swarming_url, dimensions, progress, unique, timeout, index):
  57   """Triggers a Swarming job and collects results.
  58
  59   Returns the total amount of time to run a task remotely, including all the
  60   overhead.
  61   """
  62   name = 'load-test-%d-%s' % (index, unique)
  63   start = time.time()
  64
  65   logging.info('trigger')
  66   manifest = swarming.Manifest(
  67     isolate_server='http://localhost:1',
  68     namespace='dummy-isolate',
  69     isolated_hash=1,
  70     task_name=name,
  71     shards=1,
  72     env={},
  73     dimensions=dimensions,
  74     working_dir=None,
  75     deadline=3600,
  76     verbose=False,
  77     profile=False,
  78     priority=100)
  79   # TODO(maruel): Make output size configurable.
  80   # TODO(maruel): Make number of shards configurable.
  81   output_size = 100
  82   cmd = ['python', '-c', 'print(\'1\'*%s)' % output_size]
  83   manifest.add_task('echo stuff', cmd)
  84   data = {'request': manifest.to_json()}
  85   response = net.url_open(swarming_url + '/test', data=data)
  86   if not response:
  87     # Failed to trigger. Return a failure.
  88     return 'failed_trigger'
  89   result = json.load(response)
  90   test_key = result['test_keys'][0].pop('test_key')
  91   assert test_key
  92   expected = {
  93     'test_case_name': name,
  94     'test_keys': [
  95       {
  96         # Old API uses harcoded config name.
  97         'config_name': 'isolated',
  98         'num_instances': 1,
  99         'instance_index': 0,
 100       },
 101     ],
 102   }
 103   if result != expected:
 104     # New API doesn't have concept of config name so it uses the task name.
 105     expected['test_keys'][0]['config_name'] = name
 106     assert result == expected, '%s\n%s' % (result, expected)
 107   progress.update_item('%5d' % index, processing=1)
 108   try:
 109     logging.info('collect')
 110     test_keys = swarming.get_task_keys(swarming_url, name)
 111     if not test_keys:
 112       return 'no_test_keys'
 113     assert test_keys == [test_key], test_keys
 114     out = [
 115       output
 116       for _index, output in swarming.yield_results(
 117           swarming_url, test_keys, timeout, None, False, None)
 118     ]
 119     if not out:
 120       return 'no_result'
 121     out[0].pop('machine_tag')
 122     out[0].pop('machine_id')
 123     expected = [
 124       {
 125         u'config_instance_index': 0,
 126         u'exit_codes': u'0',
 127         u'num_config_instances': 1,
 128         u'output': swarming_load_test_bot.TASK_OUTPUT,
 129       },
 130     ]
 131     assert out == expected, '\n%s\n%s' % (out, expected)
 132     return time.time() - start
 133   finally:
 134     progress.update_item('%5d - done' % index, processing=-1, processed=1)
 135
 136
 137 def main():
 138   colorama.init()
 139   parser = optparse.OptionParser(description=sys.modules[__name__].__doc__)
 140   parser.add_option(
 141       '-S', '--swarming',
 142       metavar='URL', default='',
 143       help='Swarming server to use')
 144   swarming.add_filter_options(parser)
 145   parser.set_defaults(dimensions=[('os', swarming_load_test_bot.OS_NAME)])
 146
 147   group = optparse.OptionGroup(parser, 'Load generated')
 148   group.add_option(
 149       '-s', '--send-rate', type='float', default=16., metavar='RATE',
 150       help='Rate (item/s) of sending requests as a float, default: %default')
 151   group.add_option(
 152       '-D', '--duration', type='float', default=60., metavar='N',
 153       help='Duration (s) of the sending phase of the load test, '
 154            'default: %default')
 155   group.add_option(
 156       '-m', '--concurrent', type='int', default=200, metavar='N',
 157       help='Maximum concurrent on-going requests, default: %default')
 158   group.add_option(
 159       '-t', '--timeout', type='float', default=3600., metavar='N',
 160       help='Timeout to get results, default: %default')
 161   parser.add_option_group(group)
 162
 163   group = optparse.OptionGroup(parser, 'Display options')
 164   group.add_option(
 165       '--columns', type='int', default=graph.get_console_width(), metavar='N',
 166       help='For histogram display, default:%default')
 167   group.add_option(
 168       '--buckets', type='int', default=20, metavar='N',
 169       help='Number of buckets for histogram display, default:%default')
 170   parser.add_option_group(group)
 171
 172   parser.add_option(
 173       '--dump', metavar='FOO.JSON', help='Dumps to json file')
 174   parser.add_option(
 175       '-v', '--verbose', action='store_true', help='Enables logging')
 176
 177   options, args = parser.parse_args()
 178   logging.basicConfig(level=logging.INFO if options.verbose else logging.FATAL)
 179   if args:
 180     parser.error('Unsupported args: %s' % args)
 181   options.swarming = options.swarming.rstrip('/')
 182   if not options.swarming:
 183     parser.error('--swarming is required.')
 184   if options.duration <= 0:
 185     parser.error('Needs --duration > 0. 0.01 is a valid value.')
 186   swarming.process_filter_options(parser, options)
 187
 188   total = options.send_rate * options.duration
 189   print(
 190       'Sending %.1f i/s for %ds with max %d parallel requests; timeout %.1fs; '
 191       'total %d' %
 192         (options.send_rate, options.duration, options.concurrent,
 193         options.timeout, total))
 194   print('[processing/processed/todo]')
 195
 196   # This is used so there's no clash between runs and actual real usage.
 197   unique = ''.join(random.choice(string.ascii_letters) for _ in range(8))
 198   columns = [('processing', 0), ('processed', 0), ('todo', 0)]
 199   progress = threading_utils.Progress(columns)
 200   index = 0
 201   with threading_utils.ThreadPoolWithProgress(
 202       progress, 1, options.concurrent, 0) as pool:
 203     try:
 204       start = time.time()
 205       while True:
 206         duration = time.time() - start
 207         if duration > options.duration:
 208           break
 209         should_have_triggered_so_far = int(duration * options.send_rate)
 210         while index < should_have_triggered_so_far:
 211           pool.add_task(
 212               0,
 213               trigger_task,
 214               options.swarming,
 215               options.dimensions,
 216               progress,
 217               unique,
 218               options.timeout,
 219               index)
 220           progress.update_item('', todo=1)
 221           index += 1
 222           progress.print_update()
 223         time.sleep(0.01)
 224     except KeyboardInterrupt:
 225       aborted = pool.abort()
 226       progress.update_item(
 227           'Got Ctrl-C. Aborted %d unsent tasks.' % aborted,
 228           raw=True,
 229           todo=-aborted)
 230       progress.print_update()
 231     finally:
 232       # TODO(maruel): We could give up on collecting results for the on-going
 233       # tasks but that would need to be optional.
 234       progress.update_item('Getting results for on-going tasks.', raw=True)
 235       results = sorted(pool.join())
 236   progress.print_update()
 237   # At this point, progress is not used anymore.
 238   print('')
 239   print(' - Took %.1fs.' % (time.time() - start))
 240   print('')
 241   print_results(results, options.columns, options.buckets)
 242   if options.dump:
 243     with open(options.dump, 'w') as f:
 244       json.dump(results, f, separators=(',',':'))
 245   return 0
 246
 247
 248 if __name__ == '__main__':
 249   sys.exit(main())