utils/loadtest.py

#!/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):
        """Stage artifacts for a given 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:]))