server/autotest.py

   1 # Copyright 2007 Google Inc. Released under the GPL v2
   2
   3 import re, os, sys, traceback, subprocess, time, pickle, glob, tempfile
   4 import logging, getpass
   5 from autotest_lib.server import installable_object, prebuild, utils
   6 from autotest_lib.client.common_lib import base_job, log, error, autotemp
   7 from autotest_lib.client.common_lib import global_config, packages
   8 from autotest_lib.client.common_lib import utils as client_utils
   9
  10 AUTOTEST_SVN = 'svn://test.kernel.org/autotest/trunk/client'
  11 AUTOTEST_HTTP = 'http://test.kernel.org/svn/autotest/trunk/client'
  12
  13
  14 get_value = global_config.global_config.get_config_value
  15 autoserv_prebuild = get_value('AUTOSERV', 'enable_server_prebuild',
  16                               type=bool, default=False)
  17
  18
  19 class AutodirNotFoundError(Exception):
  20     """No Autotest installation could be found."""
  21
  22
  23 class BaseAutotest(installable_object.InstallableObject):
  24     """
  25     This class represents the Autotest program.
  26
  27     Autotest is used to run tests automatically and collect the results.
  28     It also supports profilers.
  29
  30     Implementation details:
  31     This is a leaf class in an abstract class hierarchy, it must
  32     implement the unimplemented methods in parent classes.
  33     """
  34
  35     def __init__(self, host=None):
  36         self.host = host
  37         self.got = False
  38         self.installed = False
  39         self.serverdir = utils.get_server_dir()
  40         super(BaseAutotest, self).__init__()
  41
  42
  43     install_in_tmpdir = False
  44     @classmethod
  45     def set_install_in_tmpdir(cls, flag):
  46         """ Sets a flag that controls whether or not Autotest should by
  47         default be installed in a "standard" directory (e.g.
  48         /home/autotest, /usr/local/autotest) or a temporary directory. """
  49         cls.install_in_tmpdir = flag
  50
  51
  52     @classmethod
  53     def get_client_autodir_paths(cls, host):
  54         return global_config.global_config.get_config_value(
  55                 'AUTOSERV', 'client_autodir_paths', type=list)
  56
  57
  58     @classmethod
  59     def get_installed_autodir(cls, host):
  60         """
  61         Find where the Autotest client is installed on the host.
  62         @returns an absolute path to an installed Autotest client root.
  63         @raises AutodirNotFoundError if no Autotest installation can be found.
  64         """
  65         autodir = host.get_autodir()
  66         if autodir:
  67             logging.debug('Using existing host autodir: %s', autodir)
  68             return autodir
  69
  70         for path in Autotest.get_client_autodir_paths(host):
  71             try:
  72                 autotest_binary = os.path.join(path, 'bin', 'autotest')
  73                 host.run('test -x %s' % utils.sh_escape(autotest_binary))
  74                 host.run('test -w %s' % utils.sh_escape(path))
  75                 logging.debug('Found existing autodir at %s', path)
  76                 return path
  77             except error.AutoservRunError:
  78                 logging.debug('%s does not exist on %s', autotest_binary,
  79                               host.hostname)
  80         raise AutodirNotFoundError
  81
  82
  83     @classmethod
  84     def get_install_dir(cls, host):
  85         """
  86         Determines the location where autotest should be installed on
  87         host. If self.install_in_tmpdir is set, it will return a unique
  88         temporary directory that autotest can be installed in. Otherwise, looks
  89         for an existing installation to use; if none is found, looks for a
  90         usable directory in the global config client_autodir_paths.
  91         """
  92         try:
  93             install_dir = cls.get_installed_autodir(host)
  94         except AutodirNotFoundError:
  95             install_dir = cls._find_installable_dir(host)
  96
  97         if cls.install_in_tmpdir:
  98             return host.get_tmp_dir(parent=install_dir)
  99         return install_dir
 100
 101
 102     @classmethod
 103     def _find_installable_dir(cls, host):
 104         client_autodir_paths = cls.get_client_autodir_paths(host)
 105         for path in client_autodir_paths:
 106             try:
 107                 host.run('mkdir -p %s' % utils.sh_escape(path))
 108                 host.run('test -w %s' % utils.sh_escape(path))
 109                 return path
 110             except error.AutoservRunError:
 111                 logging.debug('Failed to create %s', path)
 112         raise error.AutoservInstallError(
 113                 'Unable to find a place to install Autotest; tried %s' %
 114                 ', '.join(client_autodir_paths))
 115
 116
 117     def get_fetch_location(self):
 118         c = global_config.global_config
 119         repos = c.get_config_value("PACKAGES", 'fetch_location', type=list,
 120                                    default=[])
 121         repos.reverse()
 122         return repos
 123
 124
 125     def install(self, host=None, autodir=None):
 126         self._install(host=host, autodir=autodir)
 127
 128
 129     def install_full_client(self, host=None, autodir=None):
 130         self._install(host=host, autodir=autodir, use_autoserv=False,
 131                       use_packaging=False)
 132
 133
 134     def install_no_autoserv(self, host=None, autodir=None):
 135         self._install(host=host, autodir=autodir, use_autoserv=False)
 136
 137
 138     def _install_using_packaging(self, host, autodir):
 139         repos = self.get_fetch_location()
 140         if not repos:
 141             raise error.PackageInstallError("No repos to install an "
 142                                             "autotest client from")
 143         pkgmgr = packages.PackageManager(autodir, hostname=host.hostname,
 144                                          repo_urls=repos,
 145                                          do_locking=False,
 146                                          run_function=host.run,
 147                                          run_function_dargs=dict(timeout=600))
 148         # The packages dir is used to store all the packages that
 149         # are fetched on that client. (for the tests,deps etc.
 150         # too apart from the client)
 151         pkg_dir = os.path.join(autodir, 'packages')
 152         # clean up the autodir except for the packages directory
 153         host.run('cd %s && ls | grep -v "^packages$"'
 154                  ' | xargs rm -rf && rm -rf .[^.]*' % autodir)
 155         pkgmgr.install_pkg('autotest', 'client', pkg_dir, autodir,
 156                            preserve_install_dir=True)
 157         self.installed = True
 158
 159
 160     def _install_using_send_file(self, host, autodir):
 161         dirs_to_exclude = set(["tests", "site_tests", "deps", "profilers"])
 162         light_files = [os.path.join(self.source_material, f)
 163                        for f in os.listdir(self.source_material)
 164                        if f not in dirs_to_exclude]
 165         host.send_file(light_files, autodir, delete_dest=True)
 166
 167         # create empty dirs for all the stuff we excluded
 168         commands = []
 169         for path in dirs_to_exclude:
 170             abs_path = os.path.join(autodir, path)
 171             abs_path = utils.sh_escape(abs_path)
 172             commands.append("mkdir -p '%s'" % abs_path)
 173             commands.append("touch '%s'/__init__.py" % abs_path)
 174         host.run(';'.join(commands))
 175
 176
 177     def _install(self, host=None, autodir=None, use_autoserv=True,
 178                  use_packaging=True):
 179         """
 180         Install autotest.  If get() was not called previously, an
 181         attempt will be made to install from the autotest svn
 182         repository.
 183
 184         @param host A Host instance on which autotest will be installed
 185         @param autodir Location on the remote host to install to
 186         @param use_autoserv Enable install modes that depend on the client
 187             running with the autoserv harness
 188         @param use_packaging Enable install modes that use the packaging system
 189
 190         @exception AutoservError if a tarball was not specified and
 191             the target host does not have svn installed in its path
 192         """
 193         if not host:
 194             host = self.host
 195         if not self.got:
 196             self.get()
 197         host.wait_up(timeout=30)
 198         host.setup()
 199         logging.info("Installing autotest on %s", host.hostname)
 200
 201         # set up the autotest directory on the remote machine
 202         if not autodir:
 203             autodir = self.get_install_dir(host)
 204         logging.info('Using installation dir %s', autodir)
 205         host.set_autodir(autodir)
 206         host.run('mkdir -p %s' % utils.sh_escape(autodir))
 207
 208         # make sure there are no files in $AUTODIR/results
 209         results_path = os.path.join(autodir, 'results')
 210         host.run('rm -rf %s/*' % utils.sh_escape(results_path),
 211                  ignore_status=True)
 212
 213         # Fetch the autotest client from the nearest repository
 214         if use_packaging:
 215             try:
 216                 self._install_using_packaging(host, autodir)
 217                 return
 218             except (error.PackageInstallError, error.AutoservRunError,
 219                     global_config.ConfigError), e:
 220                 logging.info("Could not install autotest using the packaging "
 221                              "system: %s. Trying other methods", e)
 222
 223         # try to install from file or directory
 224         if self.source_material:
 225             c = global_config.global_config
 226             supports_autoserv_packaging = c.get_config_value(
 227                 "PACKAGES", "serve_packages_from_autoserv", type=bool)
 228             # Copy autotest recursively
 229             if supports_autoserv_packaging and use_autoserv:
 230                 self._install_using_send_file(host, autodir)
 231             else:
 232                 host.send_file(self.source_material, autodir, delete_dest=True)
 233             logging.info("Installation of autotest completed")
 234             self.installed = True
 235             return
 236
 237         # if that fails try to install using svn
 238         if utils.run('which svn').exit_status:
 239             raise error.AutoservError('svn not found on target machine: %s' %
 240                                       host.name)
 241         try:
 242             host.run('svn checkout %s %s' % (AUTOTEST_SVN, autodir))
 243         except error.AutoservRunError, e:
 244             host.run('svn checkout %s %s' % (AUTOTEST_HTTP, autodir))
 245         logging.info("Installation of autotest completed")
 246         self.installed = True
 247
 248
 249     def uninstall(self, host=None):
 250         """
 251         Uninstall (i.e. delete) autotest. Removes the autotest client install
 252         from the specified host.
 253
 254         @params host a Host instance from which the client will be removed
 255         """
 256         if not self.installed:
 257             return
 258         if not host:
 259             host = self.host
 260         autodir = host.get_autodir()
 261         if not autodir:
 262             return
 263
 264         # perform the actual uninstall
 265         host.run("rm -rf %s" % utils.sh_escape(autodir), ignore_status=True)
 266         host.set_autodir(None)
 267         self.installed = False
 268
 269
 270     def get(self, location=None):
 271         if not location:
 272             location = os.path.join(self.serverdir, '../client')
 273             location = os.path.abspath(location)
 274         # If there's stuff run on our client directory already, it
 275         # can cause problems. Try giving it a quick clean first.
 276         cwd = os.getcwd()
 277         os.chdir(location)
 278         try:
 279             utils.system('tools/make_clean', ignore_status=True)
 280         finally:
 281             os.chdir(cwd)
 282         super(BaseAutotest, self).get(location)
 283         self.got = True
 284
 285
 286     def run(self, control_file, results_dir='.', host=None, timeout=None,
 287             tag=None, parallel_flag=False, background=False,
 288             client_disconnect_timeout=None):
 289         """
 290         Run an autotest job on the remote machine.
 291
 292         @param control_file: An open file-like-obj of the control file.
 293         @param results_dir: A str path where the results should be stored
 294                 on the local filesystem.
 295         @param host: A Host instance on which the control file should
 296                 be run.
 297         @param timeout: Maximum number of seconds to wait for the run or None.
 298         @param tag: Tag name for the client side instance of autotest.
 299         @param parallel_flag: Flag set when multiple jobs are run at the
 300                 same time.
 301         @param background: Indicates that the client should be launched as
 302                 a background job; the code calling run will be responsible
 303                 for monitoring the client and collecting the results.
 304         @param client_disconnect_timeout: Seconds to wait for the remote host
 305                 to come back after a reboot. Defaults to the host setting for
 306                 DEFAULT_REBOOT_TIMEOUT.
 307
 308         @raises AutotestRunError: If there is a problem executing
 309                 the control file.
 310         """
 311         host = self._get_host_and_setup(host)
 312         results_dir = os.path.abspath(results_dir)
 313
 314         if client_disconnect_timeout is None:
 315             client_disconnect_timeout = host.DEFAULT_REBOOT_TIMEOUT
 316
 317         if tag:
 318             results_dir = os.path.join(results_dir, tag)
 319
 320         atrun = _Run(host, results_dir, tag, parallel_flag, background)
 321         self._do_run(control_file, results_dir, host, atrun, timeout,
 322                      client_disconnect_timeout)
 323
 324
 325     def _get_host_and_setup(self, host):
 326         if not host:
 327             host = self.host
 328         if not self.installed:
 329             self.install(host)
 330
 331         host.wait_up(timeout=30)
 332         return host
 333
 334
 335     def _do_run(self, control_file, results_dir, host, atrun, timeout,
 336                 client_disconnect_timeout):
 337         try:
 338             atrun.verify_machine()
 339         except:
 340             logging.error("Verify failed on %s. Reinstalling autotest",
 341                           host.hostname)
 342             self.install(host)
 343         atrun.verify_machine()
 344         debug = os.path.join(results_dir, 'debug')
 345         try:
 346             os.makedirs(debug)
 347         except Exception:
 348             pass
 349
 350         delete_file_list = [atrun.remote_control_file,
 351                             atrun.remote_control_file + '.state',
 352                             atrun.manual_control_file,
 353                             atrun.manual_control_file + '.state']
 354         cmd = ';'.join('rm -f ' + control for control in delete_file_list)
 355         host.run(cmd, ignore_status=True)
 356
 357         tmppath = utils.get(control_file)
 358
 359         # build up the initialization prologue for the control file
 360         prologue_lines = []
 361
 362         # Add the additional user arguments
 363         prologue_lines.append("args = %r\n" % self.job.args)
 364
 365         # If the packaging system is being used, add the repository list.
 366         repos = None
 367         try:
 368             repos = self.get_fetch_location()
 369             pkgmgr = packages.PackageManager('autotest', hostname=host.hostname,
 370                                              repo_urls=repos)
 371             prologue_lines.append('job.add_repository(%s)\n' % repos)
 372         except global_config.ConfigError, e:
 373             # If repos is defined packaging is enabled so log the error
 374             if repos:
 375                 logging.error(e)
 376
 377         # on full-size installs, turn on any profilers the server is using
 378         if not atrun.background:
 379             running_profilers = host.job.profilers.add_log.iteritems()
 380             for profiler, (args, dargs) in running_profilers:
 381                 call_args = [repr(profiler)]
 382                 call_args += [repr(arg) for arg in args]
 383                 call_args += ["%s=%r" % item for item in dargs.iteritems()]
 384                 prologue_lines.append("job.profilers.add(%s)\n"
 385                                       % ", ".join(call_args))
 386         cfile = "".join(prologue_lines)
 387
 388         cfile += open(tmppath).read()
 389         open(tmppath, "w").write(cfile)
 390
 391         # Create and copy state file to remote_control_file + '.state'
 392         state_file = host.job.preprocess_client_state()
 393         host.send_file(state_file, atrun.remote_control_file + '.init.state')
 394         os.remove(state_file)
 395
 396         # Copy control_file to remote_control_file on the host
 397         host.send_file(tmppath, atrun.remote_control_file)
 398         if os.path.abspath(tmppath) != os.path.abspath(control_file):
 399             os.remove(tmppath)
 400
 401         atrun.execute_control(
 402                 timeout=timeout,
 403                 client_disconnect_timeout=client_disconnect_timeout)
 404
 405
 406     def run_timed_test(self, test_name, results_dir='.', host=None,
 407                        timeout=None, *args, **dargs):
 408         """
 409         Assemble a tiny little control file to just run one test,
 410         and run it as an autotest client-side test
 411         """
 412         if not host:
 413             host = self.host
 414         if not self.installed:
 415             self.install(host)
 416         opts = ["%s=%s" % (o[0], repr(o[1])) for o in dargs.items()]
 417         cmd = ", ".join([repr(test_name)] + map(repr, args) + opts)
 418         control = "job.run_test(%s)\n" % cmd
 419         self.run(control, results_dir, host, timeout=timeout)
 420
 421
 422     def run_test(self, test_name, results_dir='.', host=None, *args, **dargs):
 423         self.run_timed_test(test_name, results_dir, host, timeout=None,
 424                             *args, **dargs)
 425
 426
 427 class _BaseRun(object):
 428     """
 429     Represents a run of autotest control file.  This class maintains
 430     all the state necessary as an autotest control file is executed.
 431
 432     It is not intended to be used directly, rather control files
 433     should be run using the run method in Autotest.
 434     """
 435     def __init__(self, host, results_dir, tag, parallel_flag, background):
 436         self.host = host
 437         self.results_dir = results_dir
 438         self.env = host.env
 439         self.tag = tag
 440         self.parallel_flag = parallel_flag
 441         self.background = background
 442         self.autodir = Autotest.get_installed_autodir(self.host)
 443         control = os.path.join(self.autodir, 'control')
 444         if tag:
 445             control += '.' + tag
 446         self.manual_control_file = control
 447         self.remote_control_file = control + '.autoserv'
 448         self.config_file = os.path.join(self.autodir, 'global_config.ini')
 449
 450
 451     def verify_machine(self):
 452         binary = os.path.join(self.autodir, 'bin/autotest')
 453         try:
 454             self.host.run('ls %s > /dev/null 2>&1' % binary)
 455         except:
 456             raise error.AutoservInstallError(
 457                 "Autotest does not appear to be installed")
 458
 459         if not self.parallel_flag:
 460             tmpdir = os.path.join(self.autodir, 'tmp')
 461             download = os.path.join(self.autodir, 'tests/download')
 462             self.host.run('umount %s' % tmpdir, ignore_status=True)
 463             self.host.run('umount %s' % download, ignore_status=True)
 464
 465
 466     def get_base_cmd_args(self, section):
 467         args = ['--verbose']
 468         if section > 0:
 469             args.append('-c')
 470         if self.tag:
 471             args.append('-t %s' % self.tag)
 472         if self.host.job.use_external_logging():
 473             args.append('-l')
 474         if self.host.hostname:
 475             args.append('--hostname=%s' % self.host.hostname)
 476         args.append('--user=%s' % self.host.job.user)
 477
 478         args.append(self.remote_control_file)
 479         return args
 480
 481
 482     def get_background_cmd(self, section):
 483         cmd = ['nohup', os.path.join(self.autodir, 'bin/autotest_client')]
 484         cmd += self.get_base_cmd_args(section)
 485         cmd += ['>/dev/null', '2>/dev/null', '&']
 486         return ' '.join(cmd)
 487
 488
 489     def get_daemon_cmd(self, section, monitor_dir):
 490         cmd = ['nohup', os.path.join(self.autodir, 'bin/autotestd'),
 491                monitor_dir, '-H autoserv']
 492         cmd += self.get_base_cmd_args(section)
 493         cmd += ['>/dev/null', '2>/dev/null', '&']
 494         return ' '.join(cmd)
 495
 496
 497     def get_monitor_cmd(self, monitor_dir, stdout_read, stderr_read):
 498         cmd = [os.path.join(self.autodir, 'bin', 'autotestd_monitor'),
 499                monitor_dir, str(stdout_read), str(stderr_read)]
 500         return ' '.join(cmd)
 501
 502
 503     def get_client_log(self):
 504         """Find what the "next" client.* prefix should be
 505
 506         @returns A string of the form client.INTEGER that should be prefixed
 507             to all client debug log files.
 508         """
 509         max_digit = -1
 510         debug_dir = os.path.join(self.results_dir, 'debug')
 511         client_logs = glob.glob(os.path.join(debug_dir, 'client.*.*'))
 512         for log in client_logs:
 513             _, number, _ = log.split('.', 2)
 514             if number.isdigit():
 515                 max_digit = max(max_digit, int(number))
 516         return 'client.%d' % (max_digit + 1)
 517
 518
 519     def copy_client_config_file(self, client_log_prefix=None):
 520         """
 521         Create and copy the client config file based on the server config.
 522
 523         @param client_log_prefix: Optional prefix to prepend to log files.
 524         """
 525         client_config_file = self._create_client_config_file(client_log_prefix)
 526         self.host.send_file(client_config_file, self.config_file)
 527         os.remove(client_config_file)
 528
 529
 530     def _create_client_config_file(self, client_log_prefix=None):
 531         """
 532         Create a temporary file with the [CLIENT] section configuration values
 533         taken from the server global_config.ini.
 534
 535         @param client_log_prefix: Optional prefix to prepend to log files.
 536
 537         @return: Path of the temporary file generated.
 538         """
 539         config = global_config.global_config.get_section_values('CLIENT')
 540         if client_log_prefix:
 541             config.set('CLIENT', 'default_logging_name', client_log_prefix)
 542         return self._create_aux_file(config.write)
 543
 544
 545     def _create_aux_file(self, func, *args):
 546         """
 547         Creates a temporary file and writes content to it according to a
 548         content creation function. The file object is appended to *args, which
 549         is then passed to the content creation function
 550
 551         @param func: Function that will be used to write content to the
 552                 temporary file.
 553         @param *args: List of parameters that func takes.
 554         @return: Path to the temporary file that was created.
 555         """
 556         fd, path = tempfile.mkstemp(dir=self.host.job.tmpdir)
 557         aux_file = os.fdopen(fd, "w")
 558         try:
 559             list_args = list(args)
 560             list_args.append(aux_file)
 561             func(*list_args)
 562         finally:
 563             aux_file.close()
 564         return path
 565
 566
 567     @staticmethod
 568     def is_client_job_finished(last_line):
 569         return bool(re.match(r'^END .*\t----\t----\t.*$', last_line))
 570
 571
 572     @staticmethod
 573     def is_client_job_rebooting(last_line):
 574         return bool(re.match(r'^\t*GOOD\t----\treboot\.start.*$', last_line))
 575
 576
 577     def log_unexpected_abort(self, stderr_redirector):
 578         stderr_redirector.flush_all_buffers()
 579         msg = "Autotest client terminated unexpectedly"
 580         self.host.job.record("END ABORT", None, None, msg)
 581
 582
 583     def _execute_in_background(self, section, timeout):
 584         full_cmd = self.get_background_cmd(section)
 585         devnull = open(os.devnull, "w")
 586
 587         self.copy_client_config_file(self.get_client_log())
 588
 589         self.host.job.push_execution_context(self.results_dir)
 590         try:
 591             result = self.host.run(full_cmd, ignore_status=True,
 592                                    timeout=timeout,
 593                                    stdout_tee=devnull,
 594                                    stderr_tee=devnull)
 595         finally:
 596             self.host.job.pop_execution_context()
 597
 598         return result
 599
 600
 601     @staticmethod
 602     def _strip_stderr_prologue(stderr):
 603         """Strips the 'standard' prologue that get pre-pended to every
 604         remote command and returns the text that was actually written to
 605         stderr by the remote command."""
 606         stderr_lines = stderr.split("\n")[1:]
 607         if not stderr_lines:
 608             return ""
 609         elif stderr_lines[0].startswith("NOTE: autotestd_monitor"):
 610             del stderr_lines[0]
 611         return "\n".join(stderr_lines)
 612
 613
 614     def _execute_daemon(self, section, timeout, stderr_redirector,
 615                         client_disconnect_timeout):
 616         monitor_dir = self.host.get_tmp_dir()
 617         daemon_cmd = self.get_daemon_cmd(section, monitor_dir)
 618
 619         # grab the location for the server-side client log file
 620         client_log_prefix = self.get_client_log()
 621         client_log_path = os.path.join(self.results_dir, 'debug',
 622                                        client_log_prefix + '.log')
 623         client_log = open(client_log_path, 'w', 0)
 624         self.copy_client_config_file(client_log_prefix)
 625
 626         stdout_read = stderr_read = 0
 627         self.host.job.push_execution_context(self.results_dir)
 628         try:
 629             self.host.run(daemon_cmd, ignore_status=True, timeout=timeout)
 630             disconnect_warnings = []
 631             while True:
 632                 monitor_cmd = self.get_monitor_cmd(monitor_dir, stdout_read,
 633                                                    stderr_read)
 634                 try:
 635                     result = self.host.run(monitor_cmd, ignore_status=True,
 636                                            timeout=timeout,
 637                                            stdout_tee=client_log,
 638                                            stderr_tee=stderr_redirector)
 639                 except error.AutoservRunError, e:
 640                     result = e.result_obj
 641                     result.exit_status = None
 642                     disconnect_warnings.append(e.description)
 643
 644                     stderr_redirector.log_warning(
 645                         "Autotest client was disconnected: %s" % e.description,
 646                         "NETWORK")
 647                 except error.AutoservSSHTimeout:
 648                     result = utils.CmdResult(monitor_cmd, "", "", None, 0)
 649                     stderr_redirector.log_warning(
 650                         "Attempt to connect to Autotest client timed out",
 651                         "NETWORK")
 652
 653                 stdout_read += len(result.stdout)
 654                 stderr_read += len(self._strip_stderr_prologue(result.stderr))
 655
 656                 if result.exit_status is not None:
 657                     return result
 658                 elif not self.host.wait_up(client_disconnect_timeout):
 659                     raise error.AutoservSSHTimeout(
 660                         "client was disconnected, reconnect timed out")
 661         finally:
 662             client_log.close()
 663             self.host.job.pop_execution_context()
 664
 665
 666     def execute_section(self, section, timeout, stderr_redirector,
 667                         client_disconnect_timeout):
 668         logging.info("Executing %s/bin/autotest %s/control phase %d",
 669                      self.autodir, self.autodir, section)
 670
 671         if self.background:
 672             result = self._execute_in_background(section, timeout)
 673         else:
 674             result = self._execute_daemon(section, timeout, stderr_redirector,
 675                                           client_disconnect_timeout)
 676
 677         last_line = stderr_redirector.last_line
 678
 679         # check if we failed hard enough to warrant an exception
 680         if result.exit_status == 1:
 681             err = error.AutotestRunError("client job was aborted")
 682         elif not self.background and not result.stderr:
 683             err = error.AutotestRunError(
 684                 "execute_section %s failed to return anything\n"
 685                 "stdout:%s\n" % (section, result.stdout))
 686         else:
 687             err = None
 688
 689         # log something if the client failed AND never finished logging
 690         if err and not self.is_client_job_finished(last_line):
 691             self.log_unexpected_abort(stderr_redirector)
 692
 693         if err:
 694             raise err
 695         else:
 696             return stderr_redirector.last_line
 697
 698
 699     def _wait_for_reboot(self, old_boot_id):
 700         logging.info("Client is rebooting")
 701         logging.info("Waiting for client to halt")
 702         if not self.host.wait_down(self.host.WAIT_DOWN_REBOOT_TIMEOUT,
 703                                    old_boot_id=old_boot_id):
 704             err = "%s failed to shutdown after %d"
 705             err %= (self.host.hostname, self.host.WAIT_DOWN_REBOOT_TIMEOUT)
 706             raise error.AutotestRunError(err)
 707         logging.info("Client down, waiting for restart")
 708         if not self.host.wait_up(self.host.DEFAULT_REBOOT_TIMEOUT):
 709             # since reboot failed
 710             # hardreset the machine once if possible
 711             # before failing this control file
 712             warning = "%s did not come back up, hard resetting"
 713             warning %= self.host.hostname
 714             logging.warning(warning)
 715             try:
 716                 self.host.hardreset(wait=False)
 717             except (AttributeError, error.AutoservUnsupportedError):
 718                 warning = "Hard reset unsupported on %s"
 719                 warning %= self.host.hostname
 720                 logging.warning(warning)
 721             raise error.AutotestRunError("%s failed to boot after %ds" %
 722                                          (self.host.hostname,
 723                                           self.host.DEFAULT_REBOOT_TIMEOUT))
 724         self.host.reboot_followup()
 725
 726
 727     def execute_control(self, timeout=None, client_disconnect_timeout=None):
 728         if not self.background:
 729             collector = log_collector(self.host, self.tag, self.results_dir)
 730             hostname = self.host.hostname
 731             remote_results = collector.client_results_dir
 732             local_results = collector.server_results_dir
 733             self.host.job.add_client_log(hostname, remote_results,
 734                                          local_results)
 735             job_record_context = self.host.job.get_record_context()
 736
 737         section = 0
 738         start_time = time.time()
 739
 740         logger = client_logger(self.host, self.tag, self.results_dir)
 741         try:
 742             while not timeout or time.time() < start_time + timeout:
 743                 if timeout:
 744                     section_timeout = start_time + timeout - time.time()
 745                 else:
 746                     section_timeout = None
 747                 boot_id = self.host.get_boot_id()
 748                 last = self.execute_section(section, section_timeout,
 749                                             logger, client_disconnect_timeout)
 750                 if self.background:
 751                     return
 752                 section += 1
 753                 if self.is_client_job_finished(last):
 754                     logging.info("Client complete")
 755                     return
 756                 elif self.is_client_job_rebooting(last):
 757                     try:
 758                         self._wait_for_reboot(boot_id)
 759                     except error.AutotestRunError, e:
 760                         self.host.job.record("ABORT", None, "reboot", str(e))
 761                         self.host.job.record("END ABORT", None, None, str(e))
 762                         raise
 763                     continue
 764
 765                 # if we reach here, something unexpected happened
 766                 self.log_unexpected_abort(logger)
 767
 768                 # give the client machine a chance to recover from a crash
 769                 self.host.wait_up(self.host.HOURS_TO_WAIT_FOR_RECOVERY * 3600)
 770                 msg = ("Aborting - unexpected final status message from "
 771                        "client on %s: %s\n") % (self.host.hostname, last)
 772                 raise error.AutotestRunError(msg)
 773         finally:
 774             logger.close()
 775             if not self.background:
 776                 collector.collect_client_job_results()
 777                 collector.remove_redundant_client_logs()
 778                 state_file = os.path.basename(self.remote_control_file
 779                                               + '.state')
 780                 state_path = os.path.join(self.results_dir, state_file)
 781                 self.host.job.postprocess_client_state(state_path)
 782                 self.host.job.remove_client_log(hostname, remote_results,
 783                                                 local_results)
 784                 job_record_context.restore()
 785
 786         # should only get here if we timed out
 787         assert timeout
 788         raise error.AutotestTimeoutError()
 789
 790
 791 class log_collector(object):
 792     def __init__(self, host, client_tag, results_dir):
 793         self.host = host
 794         if not client_tag:
 795             client_tag = "default"
 796         self.client_results_dir = os.path.join(host.get_autodir(), "results",
 797                                                client_tag)
 798         self.server_results_dir = results_dir
 799
 800
 801     def collect_client_job_results(self):
 802         """ A method that collects all the current results of a running
 803         client job into the results dir. By default does nothing as no
 804         client job is running, but when running a client job you can override
 805         this with something that will actually do something. """
 806
 807         # make an effort to wait for the machine to come up
 808         try:
 809             self.host.wait_up(timeout=30)
 810         except error.AutoservError:
 811             # don't worry about any errors, we'll try and
 812             # get the results anyway
 813             pass
 814
 815         # Copy all dirs in default to results_dir
 816         try:
 817             self.host.get_file(self.client_results_dir + '/',
 818                                self.server_results_dir, preserve_symlinks=True)
 819         except Exception:
 820             # well, don't stop running just because we couldn't get logs
 821             e_msg = "Unexpected error copying test result logs, continuing ..."
 822             logging.error(e_msg)
 823             traceback.print_exc(file=sys.stdout)
 824
 825
 826     def remove_redundant_client_logs(self):
 827         """Remove client.*.log files in favour of client.*.DEBUG files."""
 828         debug_dir = os.path.join(self.server_results_dir, 'debug')
 829         debug_files = [f for f in os.listdir(debug_dir)
 830                        if re.search(r'^client\.\d+\.DEBUG$', f)]
 831         for debug_file in debug_files:
 832             log_file = debug_file.replace('DEBUG', 'log')
 833             log_file = os.path.join(debug_dir, log_file)
 834             if os.path.exists(log_file):
 835                 os.remove(log_file)
 836
 837
 838 # a file-like object for catching stderr from an autotest client and
 839 # extracting status logs from it
 840 class client_logger(object):
 841     """Partial file object to write to both stdout and
 842     the status log file.  We only implement those methods
 843     utils.run() actually calls.
 844     """
 845     status_parser = re.compile(r"^AUTOTEST_STATUS:([^:]*):(.*)$")
 846     test_complete_parser = re.compile(r"^AUTOTEST_TEST_COMPLETE:(.*)$")
 847     fetch_package_parser = re.compile(
 848         r"^AUTOTEST_FETCH_PACKAGE:([^:]*):([^:]*):(.*)$")
 849     extract_indent = re.compile(r"^(\t*).*$")
 850     extract_timestamp = re.compile(r".*\ttimestamp=(\d+)\t.*$")
 851
 852     def __init__(self, host, tag, server_results_dir):
 853         self.host = host
 854         self.job = host.job
 855         self.log_collector = log_collector(host, tag, server_results_dir)
 856         self.leftover = ""
 857         self.last_line = ""
 858         self.logs = {}
 859
 860
 861     def _process_log_dict(self, log_dict):
 862         log_list = log_dict.pop("logs", [])
 863         for key in sorted(log_dict.iterkeys()):
 864             log_list += self._process_log_dict(log_dict.pop(key))
 865         return log_list
 866
 867
 868     def _process_logs(self):
 869         """Go through the accumulated logs in self.log and print them
 870         out to stdout and the status log. Note that this processes
 871         logs in an ordering where:
 872
 873         1) logs to different tags are never interleaved
 874         2) logs to x.y come before logs to x.y.z for all z
 875         3) logs to x.y come before x.z whenever y < z
 876
 877         Note that this will in general not be the same as the
 878         chronological ordering of the logs. However, if a chronological
 879         ordering is desired that one can be reconstructed from the
 880         status log by looking at timestamp lines."""
 881         log_list = self._process_log_dict(self.logs)
 882         for entry in log_list:
 883             self.job.record_entry(entry, log_in_subdir=False)
 884         if log_list:
 885             self.last_line = log_list[-1].render()
 886
 887
 888     def _process_quoted_line(self, tag, line):
 889         """Process a line quoted with an AUTOTEST_STATUS flag. If the
 890         tag is blank then we want to push out all the data we've been
 891         building up in self.logs, and then the newest line. If the
 892         tag is not blank, then push the line into the logs for handling
 893         later."""
 894         entry = base_job.status_log_entry.parse(line)
 895         if entry is None:
 896             return  # the line contains no status lines
 897         if tag == "":
 898             self._process_logs()
 899             self.job.record_entry(entry, log_in_subdir=False)
 900             self.last_line = line
 901         else:
 902             tag_parts = [int(x) for x in tag.split(".")]
 903             log_dict = self.logs
 904             for part in tag_parts:
 905                 log_dict = log_dict.setdefault(part, {})
 906             log_list = log_dict.setdefault("logs", [])
 907             log_list.append(entry)
 908
 909
 910     def _process_info_line(self, line):
 911         """Check if line is an INFO line, and if it is, interpret any control
 912         messages (e.g. enabling/disabling warnings) that it may contain."""
 913         match = re.search(r"^\t*INFO\t----\t----(.*)\t[^\t]*$", line)
 914         if not match:
 915             return   # not an INFO line
 916         for field in match.group(1).split('\t'):
 917             if field.startswith("warnings.enable="):
 918                 func = self.job.warning_manager.enable_warnings
 919             elif field.startswith("warnings.disable="):
 920                 func = self.job.warning_manager.disable_warnings
 921             else:
 922                 continue
 923             warning_type = field.split("=", 1)[1]
 924             func(warning_type)
 925
 926
 927     def _process_line(self, line):
 928         """Write out a line of data to the appropriate stream. Status
 929         lines sent by autotest will be prepended with
 930         "AUTOTEST_STATUS", and all other lines are ssh error
 931         messages."""
 932         status_match = self.status_parser.search(line)
 933         test_complete_match = self.test_complete_parser.search(line)
 934         fetch_package_match = self.fetch_package_parser.search(line)
 935         if status_match:
 936             tag, line = status_match.groups()
 937             self._process_info_line(line)
 938             self._process_quoted_line(tag, line)
 939         elif test_complete_match:
 940             self._process_logs()
 941             fifo_path, = test_complete_match.groups()
 942             try:
 943                 self.log_collector.collect_client_job_results()
 944                 self.host.run("echo A > %s" % fifo_path)
 945             except Exception:
 946                 msg = "Post-test log collection failed, continuing anyway"
 947                 logging.exception(msg)
 948         elif fetch_package_match:
 949             pkg_name, dest_path, fifo_path = fetch_package_match.groups()
 950             serve_packages = global_config.global_config.get_config_value(
 951                 "PACKAGES", "serve_packages_from_autoserv", type=bool)
 952             if serve_packages and pkg_name.endswith(".tar.bz2"):
 953                 try:
 954                     self._send_tarball(pkg_name, dest_path)
 955                 except Exception:
 956                     msg = "Package tarball creation failed, continuing anyway"
 957                     logging.exception(msg)
 958             try:
 959                 self.host.run("echo B > %s" % fifo_path)
 960             except Exception:
 961                 msg = "Package tarball installation failed, continuing anyway"
 962                 logging.exception(msg)
 963         else:
 964             logging.info(line)
 965
 966
 967     def _send_tarball(self, pkg_name, remote_dest):
 968         name, pkg_type = self.job.pkgmgr.parse_tarball_name(pkg_name)
 969         src_dirs = []
 970         if pkg_type == 'test':
 971             for test_dir in ['site_tests', 'tests']:
 972                 src_dir = os.path.join(self.job.clientdir, test_dir, name)
 973                 if os.path.exists(src_dir):
 974                     src_dirs += [src_dir]
 975                     if autoserv_prebuild:
 976                         prebuild.setup(self.job.clientdir, src_dir)
 977                     break
 978         elif pkg_type == 'profiler':
 979             src_dirs += [os.path.join(self.job.clientdir, 'profilers', name)]
 980             if autoserv_prebuild:
 981                 prebuild.setup(self.job.clientdir, src_dir)
 982         elif pkg_type == 'dep':
 983             src_dirs += [os.path.join(self.job.clientdir, 'deps', name)]
 984         elif pkg_type == 'client':
 985             return  # you must already have a client to hit this anyway
 986         else:
 987             return  # no other types are supported
 988
 989         # iterate over src_dirs until we find one that exists, then tar it
 990         for src_dir in src_dirs:
 991             if os.path.exists(src_dir):
 992                 try:
 993                     logging.info('Bundling %s into %s', src_dir, pkg_name)
 994                     temp_dir = autotemp.tempdir(unique_id='autoserv-packager',
 995                                                 dir=self.job.tmpdir)
 996                     tarball_path = self.job.pkgmgr.tar_package(
 997                         pkg_name, src_dir, temp_dir.name, " .")
 998                     self.host.send_file(tarball_path, remote_dest)
 999                 finally:
1000                     temp_dir.clean()
1001                 return
1002
1003
1004     def log_warning(self, msg, warning_type):
1005         """Injects a WARN message into the current status logging stream."""
1006         timestamp = int(time.time())
1007         if self.job.warning_manager.is_valid(timestamp, warning_type):
1008             self.job.record('WARN', None, None, msg)
1009
1010
1011     def write(self, data):
1012         # now start processing the existing buffer and the new data
1013         data = self.leftover + data
1014         lines = data.split('\n')
1015         processed_lines = 0
1016         try:
1017             # process all the buffered data except the last line
1018             # ignore the last line since we may not have all of it yet
1019             for line in lines[:-1]:
1020                 self._process_line(line)
1021                 processed_lines += 1
1022         finally:
1023             # save any unprocessed lines for future processing
1024             self.leftover = '\n'.join(lines[processed_lines:])
1025
1026
1027     def flush(self):
1028         sys.stdout.flush()
1029
1030
1031     def flush_all_buffers(self):
1032         if self.leftover:
1033             self._process_line(self.leftover)
1034             self.leftover = ""
1035         self._process_logs()
1036         self.flush()
1037
1038
1039     def close(self):
1040         self.flush_all_buffers()
1041
1042
1043 SiteAutotest = client_utils.import_site_class(
1044     __file__, "autotest_lib.server.site_autotest", "SiteAutotest",
1045     BaseAutotest)
1046
1047
1048 _SiteRun = client_utils.import_site_class(
1049     __file__, "autotest_lib.server.site_autotest", "_SiteRun", _BaseRun)
1050
1051
1052 class Autotest(SiteAutotest):
1053     pass
1054
1055
1056 class _Run(_SiteRun):
1057     pass
1058
1059
1060 class AutotestHostMixin(object):
1061     """A generic mixin to add a run_test method to classes, which will allow
1062     you to run an autotest client test on a machine directly."""
1063
1064     # for testing purposes
1065     _Autotest = Autotest
1066
1067     def run_test(self, test_name, **dargs):
1068         """Run an autotest client test on the host.
1069
1070         @param test_name: The name of the client test.
1071         @param dargs: Keyword arguments to pass to the test.
1072
1073         @returns: True if the test passes, False otherwise."""
1074         at = self._Autotest()
1075         control_file = ('result = job.run_test(%s)\n'
1076                         'job.set_state("test_result", result)\n')
1077         test_args = [repr(test_name)]
1078         test_args += ['%s=%r' % (k, v) for k, v in dargs.iteritems()]
1079         control_file %= ', '.join(test_args)
1080         at.run(control_file, host=self)
1081         return at.job.get_state('test_result', default=False)