Rename load_test to isolateserver_load_test, create swarming_load_test.

tools/swarming_load_test_bot.py

+#!/usr/bin/env python
+# Copyright 2013 The Chromium Authors. All rights reserved.
+# Use of this source code is governed by a BSD-style license that can be
+# found in the LICENSE file.
+
+"""Triggers a ton of fake jobs to test its handling under high load.
+
+Generates an histogram with the latencies to process the tasks and number of
+retries.
+"""
+
+import hashlib
+import json
+import logging
+import optparse
+import os
+import Queue
+import socket
+import StringIO
+import sys
+import threading
+import time
+import zipfile
+
+ROOT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+
+sys.path.insert(0, ROOT_DIR)
+
+from third_party import colorama
+from third_party.requests.packages import urllib3
+
+from utils import graph
+from utils import net
+from utils import threading_utils
+
+# Line too long (NN/80)
+# pylint: disable=C0301
+
+
+def print_results(results, columns, buckets):
+  delays = [i for i in results if isinstance(i, float)]
+  failures = [i for i in results if not isinstance(i, float)]
+
+  print('%sDELAYS%s:' % (colorama.Fore.RED, colorama.Fore.RESET))
+  graph.print_histogram(
+      graph.generate_histogram(delays, buckets), columns, ' %.3f')
+  print('')
+  print('Total items  : %d' % len(results))
+  average = 0
+  if delays:
+    average = sum(delays)/ len(delays)
+  print('Average delay: %s' % graph.to_units(average))
+  #print('Average overhead: %s' % graph.to_units(total_size / len(sizes)))

[csharp, 2013/10/02 21:56:24]

Remove or uncomment

[M-A Ruel, 2013/10/02 23:44:25]

Done.

+  print('')
+  if failures:

[csharp, 2013/10/02 21:56:24]

Nit: new line above

[M-A Ruel, 2013/10/02 23:44:25]

Done.

+    print('%sEVENTS%s:' % (colorama.Fore.RED, colorama.Fore.RESET))
+    values = {}
+    for f in failures:
+      values.setdefault(f, 0)
+      values[f] += 1
+    graph.print_histogram(values, columns, ' %s')
+    print('')
+
+
+def calculate_version(url):

[csharp, 2013/10/02 21:56:24]

Neat way of tricking the server

+  """Retrieves the swarm_bot code and returns the SHA-1 for it."""
+  # Cannot use url_open() since zipfile requires .seek().
+  archive = zipfile.ZipFile(StringIO.StringIO(net.url_read(url)))
+  # See
+  # https://code.google.com/p/swarming/source/browse/src/common/version.py?repo=swarming-server
+  files = (
+    'slave_machine.py',
+    'swarm_bot/local_test_runner.py',
+    'common/__init__.py',
+    'common/swarm_constants.py',
+    'common/version.py',
+    'common/test_request_message.py',
+    'common/url_helper.py',
+  )
+  d = hashlib.sha1()
+  for f in files:
+    d.update(archive.read(f))
+  return d.hexdigest()
+
+
+class FakeSwarmBot(object):
+  """This is a Fake swarm_bot implementation simulating it is running AIX.
+
+  If someones fires up a real AIX slave, well, sorry.

[csharp, 2013/10/02 21:56:24]

But as the greatest OS ever, shouldn't this be a big concern?

[M-A Ruel, 2013/10/02 23:44:25]

We can fix that once it becomes a problem. I could rename to Comodore64 if you
prefer, since there's not python for Comodore.

+
+  It polls for job, acts as if it was processing them and return the fake
+  result.
+  """
+  def __init__(
+      self, swarming_url, index, progress, duration, ping, events,
+      kill_event):
+    self._lock = threading.Lock()
+    self._swarming = swarming_url
+    self._index = index
+    self._progress = progress
+    self._duration = duration
+    self._ping = ping
+    self._events = events
+    self._kill_event = kill_event
+
+    # See
+    # https://code.google.com/p/swarming/source/browse/src/swarm_bot/slave_machine.py?repo=swarming-server
+    # and
+    # https://chromium.googlesource.com/chromium/tools/build.git/+/master/scripts/tools/swarm_bootstrap/swarm_bootstrap.py
+    # for more details.
+    self._attributes = {
+      'dimensions': {
+        # Use improbable values to reduce the chance of interferring with real
+        # slaves.
+        'bits': '36',
+        'machine': os.uname()[4] + '-experimental',
+        'os': ['AIX'],
+      },
+      # Use an impossible hostname.
+      'id': '%s-%d' % (socket.getfqdn().lower(), index),
+      'try_count': 0,
+      'tag': '%s-%d' % (socket.getfqdn().lower(), index),
+      # Wait for UpdateSlave RPC to be able to calculate the proper SHA-1.
+      'version': '0' * 40,
+    }
+
+    self._thread = threading.Thread(target=self._run, name='bot%d' % index)
+    self._thread.daemon = True
+    self._thread.start()
+
+  def join(self):
+    self._thread.join()
+
+  def is_alive(self):
+    return self._thread.is_alive()
+
+  def _run(self):
+    try:
+      self._progress.update_item('%d alive' % self._index, bots=1)
+      while True:
+        if self._kill_event.get():
+          return
+        # Insert real code to fetch task from Swarming here.

[csharp, 2013/10/02 21:56:24]

I'm a bit confused by this comment, it kind of reads like a todo (which I assume
it isn't). Maybe just drop the "Insert real code to"

[M-A Ruel, 2013/10/02 23:44:25]

It was a todo that I forgot, removed.

+        data = {'attributes': json.dumps(self._attributes)}
+        try:
+          request = net.url_open(self._swarming + '/poll_for_test', data=data)
+        except urllib3.exceptions.ClosedPoolError:
+          # Work around an internal bug in urllib3.

[csharp, 2013/10/02 21:56:24]

If this is an internal urllib3 bug why are be storing it in the events? Can't we
just ignore it?

[M-A Ruel, 2013/10/02 23:44:25]

The upgrade to requests v2 should have fixed it, I removed this handler since I
don't recall seeing it since Vadim did the upgrade.

+          self._events.put('poll_for_test_urllib3')
+          continue
+        if request is None:
+          self._events.put('poll_for_test_empty')
+          continue
+        start = time.time()
+        try:
+          manifest = json.load(request)
+        except ValueError:
+          self._progress.update_item('Failed to poll')
+          self._events.put('poll_for_test_invalid')
+          continue
+
+        commands = [c['function'] for c in manifest.get('commands', [])]
+        if not commands:
+          # Nothing to run.
+          self._events.put('sleep')
+          time.sleep(manifest['come_back'])

[csharp, 2013/10/02 21:56:24]

Why we do we follow this parameter when we ignore the ping delay one? (I think
it is fine to do it this way, I'm just wondering why)

[M-A Ruel, 2013/10/02 23:44:25]

No specific reason. What do you think is best?

[csharp, 2013/10/03 16:08:04]

After thinking about this over night, I actually think we should obey the swarm
server and wait the correct amount of time before polling again or pinging.

I figure that the swarm server can adjust those values to try and reduce its
load, so the load test should respect those values. (although atm the ping isn't
adjusted based on load, but that might be an interesting future idea).

[Marc-Antoine Ruel (Google), 2013/10/03 16:47:11]

Done.

+          continue
+
+        if commands == ['UpdateSlave']:

[csharp, 2013/10/02 21:56:24]

Why not just call this once before creating all the fake bots, them we can just
initialize them with the correct version instead of having each one of them
"update"

[M-A Ruel, 2013/10/02 23:44:25]

Done.

+          # Calculate the proper SHA-1 and loop again.
+          self._attributes['version'] = calculate_version(
+              manifest['commands'][0]['args'])
+          self._events.put('update_slave')
+          continue
+
+        if commands != ['StoreFiles', 'RunCommands']:
+          self._progress.update_item(
+              'Unexpected RPC call %s\n%s' % (commands, manifest))
+          self._events.put('unknown_rpc')
+          break
+
+        # The normal way Swarming works is that it 'stores' a test_run.swarm
+        # file and then defer control to swarm_bot/local_test_runner.py.
+        store_cmd = manifest['commands'][0]
+        assert len(store_cmd['args']) == 1, store_cmd['args']
+        filepath, filename, test_run_content = store_cmd['args'][0]
+        assert filepath == ''
+        assert filename == 'test_run.swarm'
+        assert manifest['commands'][1] == {

[csharp, 2013/10/02 21:56:24]

I'm not sure we should be checking that the exact commands matches, I think just
checking it contains local_test_runner is probably enough.

[csharp, 2013/10/03 16:08:04]

ping?

[Marc-Antoine Ruel (Google), 2013/10/03 16:47:11]

Done.

+              u'function': u'RunCommands',
+              u'args': [
+                u'swarm_bot/local_test_runner.py', u'-f',
+                u'test_run.swarm', u'--restart_on_failure',
+              ],
+            }, manifest['commands'][1]
+        result_url = manifest['result_url']
+        test_run = json.loads(test_run_content)
+        assert result_url == test_run['result_url']
+        ping_url = test_run['ping_url']
+        self._progress.update_item('%d processing' % self._index, processing=1)
+
+        # Fake activity and send ping request every 0.5 second.
+        while True:

[csharp, 2013/10/02 21:56:24]

Since we know how many times we should execute this loop, why not use a for-loop
instead?

[M-A Ruel, 2013/10/02 23:44:25]

Because the call net.url_read() takes a indeterminate amount of time.

+          remaining = max(0, time.time() - start - self._duration)
+          if remaining > self._ping:
+            # In theory, we should use test_run['ping_delay'] but this is a load
+            # test. Make sure the server melts down.
+            result = net.url_read(ping_url)
+            assert result == 'OK'
+            remaining = max(0, time.time() - start - self._duration)
+          if not remaining:
+            break
+          time.sleep(remaining)
+
+        data = {
+          'c': test_run['configuration']['config_name'],
+          'n': test_run['test_run_name'],
+          'o': False,
+          'result_output': 'This task ran with great success',

[csharp, 2013/10/02 21:56:24]

Excellent output :)

+          's': True,
+          'x': '0',
+        }
+        result = net.url_read(manifest['result_url'], data=data)
+        self._progress.update_item(
+            '%d processed' % self._index, processing=-1, processed=1)
+        if not result:
+          self._events.put('result_url_fail')
+        else:
+          assert result == 'Successfully update the runner results.', result
+          self._events.put(time.time() - start)
+    finally:
+      self._progress.update_item('%d quit' % self._index, bots=-1)
+
+
+def main():
+  colorama.init()
+  parser = optparse.OptionParser(description=sys.modules[__name__].__doc__)
+  parser.add_option(
+      '-S', '--swarming',
+      metavar='URL', default='',
+      help='Swarming server to use')
+
+  group = optparse.OptionGroup(parser, 'Load generated')
+  group.add_option(
+      '--slaves', type='int', default=300, metavar='N',
+      help='Number of swarm bot slaves, default: %default')
+  group.add_option(
+      '-c', '--consume', type='float', default=60., metavar='N',
+      help='Duration (s) for consuming a request, default: %default')
+  group.add_option(
+      '-p', '--ping', type='float', default=0.5, metavar='N',
+      help='Ping delay (s) while consuming a request, normally, it would be in '
+           'the range of 30s but this is a load test, default: %default')
+  parser.add_option_group(group)
+
+  group = optparse.OptionGroup(parser, 'Display options')
+  group.add_option(
+      '--columns', type='int', default=graph.get_console_width(), metavar='N',
+      help='For histogram display, default:%default')
+  group.add_option(
+      '--buckets', type='int', default=20, metavar='N',
+      help='Number of buckets for histogram display, default:%default')
+  parser.add_option_group(group)
+
+  parser.add_option(
+      '--dump', metavar='FOO.JSON', help='Dumps to json file')
+  parser.add_option(
+      '-v', '--verbose', action='store_true', help='Enables logging')
+
+  options, args = parser.parse_args()
+  logging.basicConfig(level=logging.INFO if options.verbose else logging.FATAL)
+  if args:
+    parser.error('Unsupported args: %s' % args)
+  options.swarming = options.swarming.rstrip('/')
+  if not options.swarming:
+    parser.error('--swarming is required.')
+  if options.consume <= 0:
+    parser.error('Needs --consume > 0. 0.01 is a valid value.')
+
+  print(
+      'Running %d slaves, each task lasting %.1fs' % (
+        options.slaves, options.consume))
+
+  print('Ctrl-C to exit.')
+  print('[processing/processed/bots]')
+  columns = [('processing', 0), ('processed', 0), ('bots', 0)]
+  progress = threading_utils.Progress(columns)
+  events = Queue.Queue()
+  start = time.time()
+  kill_event = threading_utils.Bit()
+  slaves = [
+    FakeSwarmBot(
+      options.swarming, i, progress, options.consume, options.ping, events,
+      kill_event)
+    for i in range(options.slaves)
+  ]
+  try:
+    # Wait for all the slaves to come alive.
+    while not all(s.is_alive() for s in slaves):
+      time.sleep(0.01)
+    progress.update_item('Ready to run')
+    while slaves:
+      progress.print_update()
+      time.sleep(0.01)
+      # The slaves could be told to suicide.
+      slaves = [s for s in slaves if s.is_alive()]
+  except KeyboardInterrupt:
+    kill_event.set()
+
+  progress.update_item('Waiting for slaves to quit.', raw=True)
+  progress.update_item('')
+  while slaves:
+    progress.print_update()
+    slaves = [s for s in slaves if s.is_alive()]
+  # At this point, progress is not used anymore.
+  print('')
+  print('Ran for %.1fs.' % (time.time() - start))
+  print('')
+  results = events.queue
+  print_results(results, options.columns, options.buckets)
+  if options.dump:
+    with open(options.dump, 'w') as f:
+      json.dump(results, f, separators=(',',':'))
+  return 0
+
+
+if __name__ == '__main__':
+  sys.exit(main())