tko/parse.py

   1 #!/usr/bin/python -u
   2
   3 import os, sys, optparse, fcntl, errno, traceback, socket
   4
   5 import common
   6 from autotest_lib.client.common_lib import mail, pidfile
   7 from autotest_lib.tko import db as tko_db, utils as tko_utils, status_lib, models
   8 from autotest_lib.client.common_lib import utils
   9
  10
  11 def parse_args():
  12     # build up our options parser and parse sys.argv
  13     parser = optparse.OptionParser()
  14     parser.add_option("-m", help="Send mail for FAILED tests",
  15                       dest="mailit", action="store_true")
  16     parser.add_option("-r", help="Reparse the results of a job",
  17                       dest="reparse", action="store_true")
  18     parser.add_option("-o", help="Parse a single results directory",
  19                       dest="singledir", action="store_true")
  20     parser.add_option("-l", help=("Levels of subdirectories to include "
  21                                   "in the job name"),
  22                       type="int", dest="level", default=1)
  23     parser.add_option("-n", help="No blocking on an existing parse",
  24                       dest="noblock", action="store_true")
  25     parser.add_option("-s", help="Database server hostname",
  26                       dest="db_host", action="store")
  27     parser.add_option("-u", help="Database username", dest="db_user",
  28                       action="store")
  29     parser.add_option("-p", help="Database password", dest="db_pass",
  30                       action="store")
  31     parser.add_option("-d", help="Database name", dest="db_name",
  32                       action="store")
  33     parser.add_option("--write-pidfile",
  34                       help="write pidfile (.parser_execute)",
  35                       dest="write_pidfile", action="store_true",
  36                       default=False)
  37     options, args = parser.parse_args()
  38
  39     # we need a results directory
  40     if len(args) == 0:
  41         tko_utils.dprint("ERROR: at least one results directory must "
  42                          "be provided")
  43         parser.print_help()
  44         sys.exit(1)
  45
  46     # pass the options back
  47     return options, args
  48
  49
  50 def format_failure_message(jobname, kernel, testname, status, reason):
  51     format_string = "%-12s %-20s %-12s %-10s %s"
  52     return format_string % (jobname, kernel, testname, status, reason)
  53
  54
  55 def mailfailure(jobname, job, message):
  56     message_lines = [""]
  57     message_lines.append("The following tests FAILED for this job")
  58     message_lines.append("http://%s/results/%s" %
  59                          (socket.gethostname(), jobname))
  60     message_lines.append("")
  61     message_lines.append(format_failure_message("Job name", "Kernel",
  62                                                 "Test name", "FAIL/WARN",
  63                                                 "Failure reason"))
  64     message_lines.append(format_failure_message("=" * 8, "=" * 6, "=" * 8,
  65                                                 "=" * 8, "=" * 14))
  66     message_header = "\n".join(message_lines)
  67
  68     subject = "AUTOTEST: FAILED tests from job %s" % jobname
  69     mail.send("", job.user, "", subject, message_header + message)
  70
  71
  72 def parse_one(db, jobname, path, reparse, mail_on_failure):
  73     """
  74     Parse a single job. Optionally send email on failure.
  75     """
  76     tko_utils.dprint("\nScanning %s (%s)" % (jobname, path))
  77     old_job_idx = db.find_job(jobname)
  78     # old tests is a dict from tuple (test_name, subdir) to test_idx
  79     old_tests = {}
  80     if old_job_idx is not None:
  81         if not reparse:
  82             tko_utils.dprint("! Job is already parsed, done")
  83             return
  84
  85         raw_old_tests = db.select("test_idx,subdir,test", "tko_tests",
  86                                   {"job_idx": old_job_idx})
  87         if raw_old_tests:
  88             old_tests = dict(((test, subdir), test_idx)
  89                              for test_idx, subdir, test in raw_old_tests)
  90
  91     # look up the status version
  92     job_keyval = models.job.read_keyval(path)
  93     status_version = job_keyval.get("status_version", 0)
  94
  95     # parse out the job
  96     parser = status_lib.parser(status_version)
  97     job = parser.make_job(path)
  98     status_log = os.path.join(path, "status.log")
  99     if not os.path.exists(status_log):
 100         status_log = os.path.join(path, "status")
 101     if not os.path.exists(status_log):
 102         tko_utils.dprint("! Unable to parse job, no status file")
 103         return
 104
 105     # parse the status logs
 106     tko_utils.dprint("+ Parsing dir=%s, jobname=%s" % (path, jobname))
 107     status_lines = open(status_log).readlines()
 108     parser.start(job)
 109     tests = parser.end(status_lines)
 110
 111     # parser.end can return the same object multiple times, so filter out dups
 112     job.tests = []
 113     already_added = set()
 114     for test in tests:
 115         if test not in already_added:
 116             already_added.add(test)
 117             job.tests.append(test)
 118
 119     # try and port test_idx over from the old tests, but if old tests stop
 120     # matching up with new ones just give up
 121     if reparse and old_job_idx is not None:
 122         job.index = old_job_idx
 123         for test in job.tests:
 124             test_idx = old_tests.pop((test.testname, test.subdir), None)
 125             if test_idx is not None:
 126                 test.test_idx = test_idx
 127             else:
 128                 tko_utils.dprint("! Reparse returned new test "
 129                                  "testname=%r subdir=%r" %
 130                                  (test.testname, test.subdir))
 131         for test_idx in old_tests.itervalues():
 132             where = {'test_idx' : test_idx}
 133             db.delete('tko_iteration_result', where)
 134             db.delete('tko_iteration_attributes', where)
 135             db.delete('tko_test_attributes', where)
 136             db.delete('tko_test_labels_tests', {'test_id': test_idx})
 137             db.delete('tko_tests', where)
 138
 139     # check for failures
 140     message_lines = [""]
 141     for test in job.tests:
 142         if not test.subdir:
 143             continue
 144         tko_utils.dprint("* testname, status, reason: %s %s %s"
 145                          % (test.subdir, test.status, test.reason))
 146         if test.status in ("FAIL", "WARN"):
 147             message_lines.append(format_failure_message(
 148                 jobname, test.kernel.base, test.subdir,
 149                 test.status, test.reason))
 150     message = "\n".join(message_lines)
 151
 152     # send out a email report of failure
 153     if len(message) > 2 and mail_on_failure:
 154         tko_utils.dprint("Sending email report of failure on %s to %s"
 155                          % (jobname, job.user))
 156         mailfailure(jobname, job, message)
 157
 158     # write the job into the database
 159     db.insert_job(jobname, job)
 160
 161     # Serializing job into a binary file
 162     try:
 163         from autotest_lib.tko import tko_pb2
 164         from autotest_lib.tko import job_serializer
 165
 166         serializer = job_serializer.JobSerializer()
 167         binary_file_name = os.path.join(path, "job.serialize")
 168         serializer.serialize_to_binary(job, jobname, binary_file_name)
 169
 170         if reparse:
 171             site_export_file = "autotest_lib.tko.site_export"
 172             site_export = utils.import_site_function(__file__,
 173                                                      site_export_file,
 174                                                      "site_export",
 175                                                      _site_export_dummy)
 176             site_export(binary_file_name)
 177
 178     except ImportError:
 179         tko_utils.dprint("DEBUG: tko_pb2.py doesn't exist. Create by "
 180                          "compiling tko/tko.proto.")
 181
 182     db.commit()
 183
 184 def _site_export_dummy(binary_file_name):
 185     pass
 186
 187 def _get_job_subdirs(path):
 188     """
 189     Returns a list of job subdirectories at path. Returns None if the test
 190     is itself a job directory. Does not recurse into the subdirs.
 191     """
 192     # if there's a .machines file, use it to get the subdirs
 193     machine_list = os.path.join(path, ".machines")
 194     if os.path.exists(machine_list):
 195         subdirs = set(line.strip() for line in file(machine_list))
 196         existing_subdirs = set(subdir for subdir in subdirs
 197                                if os.path.exists(os.path.join(path, subdir)))
 198         if len(existing_subdirs) != 0:
 199             return existing_subdirs
 200
 201     # if this dir contains ONLY subdirectories, return them
 202     contents = set(os.listdir(path))
 203     contents.discard(".parse.lock")
 204     subdirs = set(sub for sub in contents if
 205                   os.path.isdir(os.path.join(path, sub)))
 206     if len(contents) == len(subdirs) != 0:
 207         return subdirs
 208
 209     # this is a job directory, or something else we don't understand
 210     return None
 211
 212
 213 def parse_leaf_path(db, path, level, reparse, mail_on_failure):
 214     job_elements = path.split("/")[-level:]
 215     jobname = "/".join(job_elements)
 216     try:
 217         db.run_with_retry(parse_one, db, jobname, path, reparse,
 218                           mail_on_failure)
 219     except Exception:
 220         traceback.print_exc()
 221
 222
 223 def parse_path(db, path, level, reparse, mail_on_failure):
 224     job_subdirs = _get_job_subdirs(path)
 225     if job_subdirs is not None:
 226         # parse status.log in current directory, if it exists. multi-machine
 227         # synchronous server side tests record output in this directory. without
 228         # this check, we do not parse these results.
 229         if os.path.exists(os.path.join(path, 'status.log')):
 230             parse_leaf_path(db, path, level, reparse, mail_on_failure)
 231         # multi-machine job
 232         for subdir in job_subdirs:
 233             jobpath = os.path.join(path, subdir)
 234             parse_path(db, jobpath, level + 1, reparse, mail_on_failure)
 235     else:
 236         # single machine job
 237         parse_leaf_path(db, path, level, reparse, mail_on_failure)
 238
 239
 240 def main():
 241     options, args = parse_args()
 242     results_dir = os.path.abspath(args[0])
 243     assert os.path.exists(results_dir)
 244
 245     pid_file_manager = pidfile.PidFileManager("parser", results_dir)
 246
 247     if options.write_pidfile:
 248         pid_file_manager.open_file()
 249
 250     try:
 251         # build up the list of job dirs to parse
 252         if options.singledir:
 253             jobs_list = [results_dir]
 254         else:
 255             jobs_list = [os.path.join(results_dir, subdir)
 256                          for subdir in os.listdir(results_dir)]
 257
 258         # build up the database
 259         db = tko_db.db(autocommit=False, host=options.db_host,
 260                        user=options.db_user, password=options.db_pass,
 261                        database=options.db_name)
 262
 263         # parse all the jobs
 264         for path in jobs_list:
 265             lockfile = open(os.path.join(path, ".parse.lock"), "w")
 266             flags = fcntl.LOCK_EX
 267             if options.noblock:
 268                 flags |= fcntl.LOCK_NB
 269             try:
 270                 fcntl.flock(lockfile, flags)
 271             except IOError, e:
 272                 # lock is not available and nonblock has been requested
 273                 if e.errno == errno.EWOULDBLOCK:
 274                     lockfile.close()
 275                     continue
 276                 else:
 277                     raise # something unexpected happened
 278             try:
 279                 parse_path(db, path, options.level, options.reparse,
 280                            options.mailit)
 281
 282             finally:
 283                 fcntl.flock(lockfile, fcntl.LOCK_UN)
 284                 lockfile.close()
 285
 286     except:
 287         pid_file_manager.close_file(1)
 288         raise
 289     else:
 290         pid_file_manager.close_file(0)
 291
 292
 293 if __name__ == "__main__":
 294     main()