#!/usr/bin/env python2
# Copyright 2017 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Load generator for devserver.
Example usage:
# Find DUTs in suites pool to test with:
atest host list -b 'pool:suites,board:BOARD' --unlocked -s Ready
# Lock DUTs:
atest host mod -l -r 'quick provision testing' DUT1 DUT2
# Create config file with DUTs to test and builds to use.
cat >config.json <<EOD
{
"BOARD": {
"duts": [
"chromeosX-rowY-rackZ-hostA",
"chromeosX-rowY-rackZ-hostB",
],
"versions": [
"auron_paine-paladin/R65-10208.0.0-rc2",
"auron_paine-paladin/R65-10208.0.0-rc3",
"auron_paine-paladin/R65-10209.0.0-rc1"
]
},
}
EOD
# Do 100 total provisions, aiming to have 10 active simultaneously.
loadtest.py $DS config.json --simultaneous 10 --total 100
# Unlock DUTs:
atest host mod -u DUT1 DUT2
"""
import collections
import datetime
import json
import random
import re
import signal
import subprocess
import sys
import time
import common
from autotest_lib.client.common_lib import time_utils
from autotest_lib.client.common_lib.cros import dev_server
from chromite.lib import commandline
from chromite.lib import cros_logging as logging
from chromite.lib import locking
from chromite.lib import parallel
# Paylods to stage.
PAYLOADS = ['quick_provision', 'stateful']
# Number of seconds between full status checks.
STATUS_POLL_SECONDS = 2
# Number of successes/failures to blacklist a DUT.
BLACKLIST_CONSECUTIVE_FAILURE = 2
BLACKLIST_TOTAL_SUCCESS = 0
BLACKLIST_TOTAL_FAILURE = 5
def get_parser():
"""Creates the argparse parser."""
parser = commandline.ArgumentParser(description=__doc__)
parser.add_argument('server', type=str, action='store',
help='Devserver to load test.')
parser.add_argument('config', type=str, action='store',
help='Path to JSON config file.'
'Config file is indexed by board with keys of '
'"duts" and "versions", each a list.')
parser.add_argument('--blacklist-consecutive', '-C', type=int,
action='store',
help=('Consecutive number of failures before '
'blacklisting DUT (default %d).') %
BLACKLIST_CONSECUTIVE_FAILURE,
default=BLACKLIST_CONSECUTIVE_FAILURE)
parser.add_argument('--blacklist-success', '-S', type=int, action='store',
help=('Total number of successes before blacklisting '
'DUT (default %d).') % BLACKLIST_TOTAL_SUCCESS,
default=BLACKLIST_TOTAL_SUCCESS)
parser.add_argument('--blacklist-total', '-T', type=int, action='store',
help=('Total number of failures before blacklisting '
'DUT (default %d).') % BLACKLIST_TOTAL_FAILURE,
default=BLACKLIST_TOTAL_FAILURE)
parser.add_argument('--boards', '-b', type=str, action='store',
help='Comma-separated list of boards to provision.')
parser.add_argument('--dryrun', '-n', action='store_true', dest='dryrun',
help='Do not attempt to provision.')
parser.add_argument('--duts', '-d', type=str, action='store',
help='Comma-separated list of duts to provision.')
parser.add_argument('--outputlog', '-l', type=str, action='store',
help='Path to append JSON entries to.')
parser.add_argument('--output', '-o', type=str, action='store',
help='Path to write JSON file to.')
parser.add_argument('--ping', '-p', action='store_true',
help='Ping DUTs and blacklist unresponsive ones.')
parser.add_argument('--simultaneous', '-s', type=int, action='store',
help='Number of simultaneous provisions to run.',
default=1)
parser.add_argument('--no-stage', action='store_false',
dest='stage', default=True,
help='Do not attempt to stage builds.')
parser.add_argument('--total', '-t', type=int, action='store',
help='Number of total provisions to run.',
default=0)
return parser
def make_entry(entry_id, name, status, start_time,
finish_time=None, parent=None, **kwargs):
"""Generate an event log entry to be stored in Cloud Datastore.
@param entry_id: A (Kind, id) tuple representing the key.
@param name: A string identifying the event
@param status: A string identifying the status of the event.
@param start_time: A datetime of the start of the event.
@param finish_time: A datetime of the finish of the event.
@param parent: A (Kind, id) tuple representing the parent key.
@return A dictionary representing the entry suitable for dumping via JSON.
"""
entry = {
'id': entry_id,
'name': name,
'status': status,
'start_time': time_utils.to_epoch_time(start_time),
}
if finish_time is not None:
entry['finish_time'] = time_utils.to_epoch_time(finish_time)
if parent is not None:
entry['parent'] = parent
return entry
class Job(object):
"""Tracks a single provision job."""
def __init__(self, ds, host_name, build_name,
entry_id=0, parent=None, board=None,
start_active=0,
force_update=False, full_update=False,
clobber_stateful=True, quick_provision=True,
ping=False, dryrun=False):
self.ds = ds
self.host_name = host_name
self.build_name = build_name
self.entry_id = ('Job', entry_id)
self.parent = parent
self.board = board
self.start_active = start_active
self.end_active = None
self.check_active_sum = 0
self.check_active_count = 0
self.start_time = datetime.datetime.now()
self.finish_time = None
self.trigger_response = None
self.ping = ping
self.pre_ping = None
self.post_ping = None
self.kwargs = {
'host_name': host_name,
'build_name': build_name,
'force_update': force_update,
'full_update': full_update,
'clobber_stateful': clobber_stateful,
'quick_provision': quick_provision,
}
if dryrun:
self.finish_time = datetime.datetime.now()
self.raised_error = None
self.success = True
self.pid = 0
else:
if self.ping:
self.pre_ping = ping_dut(self.host_name)
self.trigger_response = ds._trigger_auto_update(**self.kwargs)
def as_entry(self):
"""Generate an entry for exporting to datastore."""
entry = make_entry(self.entry_id, self.host_name,
'pass' if self.success else 'fail',
self.start_time, self.finish_time, self.parent)
entry.update({
'build_name': self.build_name,
'board': self.board,
'devserver': self.ds.hostname,
'start_active': self.start_active,
'end_active': self.end_active,
'force_update': self.kwargs['force_update'],
'full_update': self.kwargs['full_update'],
'clobber_stateful': self.kwargs['clobber_stateful'],
'quick_provision': self.kwargs['quick_provision'],
'elapsed': int(self.elapsed().total_seconds()),
'trigger_response': self.trigger_response,
'pre_ping': self.pre_ping,
'post_ping': self.post_ping,
})
if self.check_active_count:
entry['avg_active'] = (self.check_active_sum /
self.check_active_count)
return entry
def check(self, active_count):
"""Checks if a job has completed.
@param active_count: Number of active provisions at time of the check.
@return: True if the job has completed, False otherwise.
"""
if self.finish_time is not None:
return True
self.check_active_sum += active_count
self.check_active_count += 1
finished, raised_error, pid = self.ds.check_for_auto_update_finished(
self.trigger_response, wait=False, **self.kwargs)
if finished:
self.finish_time = datetime.datetime.now()
self.raised_error = raised_error
self.success = self.raised_error is None
self.pid = pid
self.end_active = active_count
if self.ping:
self.post_ping = ping_dut(self.host_name)
return finished
def elapsed(self):
"""Determine the elapsed time of the task."""
finish_time = self.finish_time or datetime.datetime.now()
return finish_time - self.start_time
class Runner(object):
"""Parallel provision load test runner."""
def __init__(self, ds, duts, config, simultaneous=1, total=0,
outputlog=None, ping=False, blacklist_consecutive=None,
blacklist_success=None, blacklist_total=None, dryrun=False):
self.ds = ds
self.duts = duts
self.config = config
self.start_time = datetime.datetime.now()
self.finish_time = None
self.simultaneous = simultaneous
self.total = total
self.outputlog = outputlog
self.ping = ping
self.blacklist_consecutive = blacklist_consecutive
self.blacklist_success = blacklist_success
self.blacklist_total = blacklist_total
self.dryrun = dryrun
self.active = []
self.started = 0
self.completed = []
# Track DUTs which have failed multiple times.
self.dut_blacklist = set()
# Track versions of each DUT to provision in order.
self.last_versions = {}
# id for the parent entry.
# TODO: This isn't the most unique.
self.entry_id = ('Runner',
int(time_utils.to_epoch_time(datetime.datetime.now())))
# ids for the job entries.
self.next_id = 0
if self.outputlog:
dump_entries_as_json([self.as_entry()], self.outputlog)
def signal_handler(self, signum, frame):
"""Signal handle to dump current status."""
logging.info('Received signal %s', signum)
if signum == signal.SIGUSR1:
now = datetime.datetime.now()
logging.info('%d active provisions, %d completed provisions, '
'%s elapsed:',
len(self.active), len(self.completed),
now - self.start_time)
for job in self.active:
logging.info(' %s -> %s, %s elapsed',
job.host_name, job.build_name,
now - job.start_time)
def as_entry(self):
"""Generate an entry for exporting to datastore."""
entry = make_entry(self.entry_id, 'Runner', 'pass',
self.start_time, self.finish_time)
entry.update({
'devserver': self.ds.hostname,
})
return entry
def get_completed_entries(self):
"""Retrieves all completed jobs as entries for datastore."""
entries = [self.as_entry()]
entries.extend([job.as_entry() for job in self.completed])
return entries
def get_next_id(self):
"""Get the next Job id."""
entry_id = self.next_id
self.next_id += 1
return entry_id
def spawn(self, host_name, build_name):
"""Spawn a single provision job."""
job = Job(self.ds, host_name, build_name,
entry_id=self.get_next_id(), parent=self.entry_id,
board=self.get_dut_board_type(host_name),
start_active=len(self.active), ping=self.ping,
dryrun=self.dryrun)
self.active.append(job)
logging.info('Provision (%d) of %s to %s started',
job.entry_id[1], job.host_name, job.build_name)
self.last_versions[host_name] = build_name
self.started += 1
def replenish(self):
"""Replenish the number of active provisions to match goals."""
while ((self.simultaneous == 0 or
len(self.active) < self.simultaneous) and
(self.total == 0 or self.started < self.total)):
host_name = self.find_idle_dut()
if host_name:
build_name = self.find_build_for_dut(host_name)
self.spawn(host_name, build_name)
elif self.simultaneous:
logging.warn('Insufficient DUTs to satisfy goal')
return False
else:
return len(self.active) > 0
return True
def check_all(self):
"""Check the status of outstanding provisions."""
still_active = []
for job in self.active:
if job.check(len(self.active)):
logging.info('Provision (%d) of %s to %s %s in %s: %s',
job.entry_id[1], job.host_name, job.build_name,
'completed' if job.success else 'failed',
job.elapsed(), job.raised_error)
entry = job.as_entry()
logging.debug(json.dumps(entry))
if self.outputlog:
dump_entries_as_json([entry], self.outputlog)
self.completed.append(job)
if self.should_blacklist(job.host_name):
logging.error('Blacklisting DUT %s', job.host_name)
self.dut_blacklist.add(job.host_name)
else:
still_active.append(job)
self.active = still_active
def should_blacklist(self, host_name):
"""Determines if a given DUT should be blacklisted."""
jobs = [job for job in self.completed if job.host_name == host_name]
total = 0
consecutive = 0
successes = 0
for job in jobs:
if not job.success:
total += 1
consecutive += 1
if ((self.blacklist_total is not None and
total >= self.blacklist_total) or
(self.blacklist_consecutive is not None and
consecutive >= self.blacklist_consecutive)):
return True
else:
successes += 1
if (self.blacklist_success is not None and
successes >= self.blacklist_success):
return True
consecutive = 0
return False
def find_idle_dut(self):
"""Find an idle DUT to provision.."""
active_duts = {job.host_name for job in self.active}
idle_duts = [d for d in self.duts
if d not in active_duts | self.dut_blacklist]
return random.choice(idle_duts) if len(idle_duts) else None
def get_dut_board_type(self, host_name):
"""Determine the board type of a DUT."""
return self.duts[host_name]
def get_board_versions(self, board):
"""Determine the versions to provision for a board."""
return self.config[board]['versions']
def find_build_for_dut(self, host_name):
"""Determine a build to provision on a DUT."""
board = self.get_dut_board_type(host_name)
versions = self.get_board_versions(board)
last_version = self.last_versions.get(host_name)
try:
last_index = versions.index(last_version)
except ValueError:
return versions[0]
return versions[(last_index + 1) % len(versions)]
def stage(self, build):
logging.debug('Staging %s', build)
self.ds.stage_artifacts(build, PAYLOADS)
def stage_all(self):
"""Stage all necessary artifacts."""
boards = set(self.duts.values())
logging.info('Staging for %d boards', len(boards))
funcs = []
for board in boards:
for build in self.get_board_versions(board):
funcs.append(lambda build_=build: self.stage(build_))
parallel.RunParallelSteps(funcs)
def loop(self):
"""Run the main provision loop."""
# Install a signal handler for status updates.
old_handler = signal.signal(signal.SIGUSR1, self.signal_handler)
signal.siginterrupt(signal.SIGUSR1, False)
try:
while True:
self.check_all()
if self.total != 0 and len(self.completed) >= self.total:
break
if not self.replenish() and len(self.active) == 0:
logging.error('Unable to replenish with no active '
'provisions')
return False
logging.debug('%d provisions active', len(self.active))
time.sleep(STATUS_POLL_SECONDS)
return True
except KeyboardInterrupt:
return False
finally:
self.finish_time = datetime.datetime.now()
# Clean up signal handler.
signal.signal(signal.SIGUSR1, old_handler)
def elapsed(self):
"""Determine the elapsed time of the task."""
finish_time = self.finish_time or datetime.datetime.now()
return finish_time - self.start_time
def dump_entries_as_json(entries, output_file):
"""Dump event log entries as json to a file.
@param entries: A list of event log entries to dump.
@param output_file: The file to write to.
"""
# Write the entries out as JSON.
logging.debug('Dumping %d entries' % len(entries))
for e in entries:
json.dump(e, output_file, sort_keys=True)
output_file.write('\n')
output_file.flush()
def ping_dut(hostname):
"""Checks if a host is responsive to pings."""
if re.match('^chromeos\d+-row\d+-rack\d+-host\d+$', hostname):
hostname += '.cros'
response = subprocess.call(["/bin/ping", "-c1", "-w2", hostname],
stdout=subprocess.PIPE)
return response == 0
def main(argv):
"""Load generator for a devserver."""
parser = get_parser()
options = parser.parse_args(argv)
# Parse devserver.
if options.server:
if re.match(r'^https?://', options.server):
server = options.server
else:
server = 'http://%s/' % options.server
ds = dev_server.ImageServer(server)
else:
parser.print_usage()
logging.error('Must specify devserver')
sys.exit(1)
# Parse config file and determine master list of duts and their board type,
# filtering by board type if specified.
duts = {}
if options.config:
with open(options.config, 'r') as f:
config = json.load(f)
boards = (options.boards.split(',')
if options.boards else config.keys())
duts_specified = (set(options.duts.split(','))
if options.duts else None)
for board in boards:
duts.update({dut: board for dut in config[board]['duts']
if duts_specified is None or
dut in duts_specified})
logging.info('Config file %s: %d boards, %d duts',
options.config, len(boards), len(duts))
else:
parser.print_usage()
logging.error('Must specify config file')
sys.exit(1)
if options.ping:
logging.info('Performing ping tests')
duts_alive = {}
for dut, board in duts.items():
if ping_dut(dut):
duts_alive[dut] = board
else:
logging.error('Ignoring DUT %s (%s) for failing initial '
'ping check', dut, board)
duts = duts_alive
logging.info('After ping tests: %d boards, %d duts', len(boards),
len(duts))
# Set up the test runner and stage all the builds.
outputlog = open(options.outputlog, 'a') if options.outputlog else None
runner = Runner(ds, duts, config,
simultaneous=options.simultaneous, total=options.total,
outputlog=outputlog, ping=options.ping,
blacklist_consecutive=options.blacklist_consecutive,
blacklist_success=options.blacklist_success,
blacklist_total=options.blacklist_total,
dryrun=options.dryrun)
if options.stage:
runner.stage_all()
# Run all the provisions.
with locking.FileLock(options.config, blocking=True).lock():
completed = runner.loop()
logging.info('%s in %s', 'Completed' if completed else 'Interrupted',
runner.elapsed())
# Write all entries as JSON.
entries = runner.get_completed_entries()
if options.output:
with open(options.output, 'w') as f:
dump_entries_as_json(entries, f)
else:
dump_entries_as_json(entries, sys.stdout)
logging.info('Summary: %s',
dict(collections.Counter([e['status'] for e in entries
if e['name'] != 'Runner'])))
# List blacklisted DUTs.
if runner.dut_blacklist:
logging.warn('Blacklisted DUTs:')
for host_name in runner.dut_blacklist:
logging.warn(' %s', host_name)
if __name__ == '__main__':
sys.exit(main(sys.argv[1:]))