tko/parse.py

#!/usr/bin/python -u

import os, sys, optparse, fcntl, errno, traceback, socket

import common
from autotest_lib.client.common_lib import mail, pidfile
from autotest_lib.tko import db as tko_db, utils as tko_utils, status_lib, models
from autotest_lib.client.common_lib import utils


def parse_args():
    # build up our options parser and parse sys.argv
    parser = optparse.OptionParser()
    parser.add_option("-m", help="Send mail for FAILED tests",
                      dest="mailit", action="store_true")
    parser.add_option("-r", help="Reparse the results of a job",
                      dest="reparse", action="store_true")
    parser.add_option("-o", help="Parse a single results directory",
                      dest="singledir", action="store_true")
    parser.add_option("-l", help=("Levels of subdirectories to include "
                                  "in the job name"),
                      type="int", dest="level", default=1)
    parser.add_option("-n", help="No blocking on an existing parse",
                      dest="noblock", action="store_true")
    parser.add_option("-s", help="Database server hostname",
                      dest="db_host", action="store")
    parser.add_option("-u", help="Database username", dest="db_user",
                      action="store")
    parser.add_option("-p", help="Database password", dest="db_pass",
                      action="store")
    parser.add_option("-d", help="Database name", dest="db_name",
                      action="store")
    parser.add_option("-P", help="Run site post-processing",
                      dest="site_do_post", action="store_true", default=False)
    parser.add_option("--write-pidfile",
                      help="write pidfile (.parser_execute)",
                      dest="write_pidfile", action="store_true",
                      default=False)
    options, args = parser.parse_args()

    # we need a results directory
    if len(args) == 0:
        tko_utils.dprint("ERROR: at least one results directory must "
                         "be provided")
        parser.print_help()
        sys.exit(1)

    # pass the options back
    return options, args


def format_failure_message(jobname, kernel, testname, status, reason):
    format_string = "%-12s %-20s %-12s %-10s %s"
    return format_string % (jobname, kernel, testname, status, reason)


def mailfailure(jobname, job, message):
    message_lines = [""]
    message_lines.append("The following tests FAILED for this job")
    message_lines.append("http://%s/results/%s" %
                         (socket.gethostname(), jobname))
    message_lines.append("")
    message_lines.append(format_failure_message("Job name", "Kernel",
                                                "Test name", "FAIL/WARN",
                                                "Failure reason"))
    message_lines.append(format_failure_message("=" * 8, "=" * 6, "=" * 8,
                                                "=" * 8, "=" * 14))
    message_header = "\n".join(message_lines)

    subject = "AUTOTEST: FAILED tests from job %s" % jobname
    mail.send("", job.user, "", subject, message_header + message)


def find_old_tests(db, job_idx):
    """
    Given a job index, return a list of all the test objects associated with
    it in the database, including labels, but excluding other "complex"
    data (attributes, iteration data, kernels).
    """
    raw_tests = db.select("test_idx,subdir,test,started_time,finished_time",
                          "tests", {"job_idx": job_idx})
    if not raw_tests:
        return []

    test_ids = ", ".join(str(raw_test[0]) for raw_test in raw_tests)

    labels = db.select("test_id, testlabel_id", "test_labels_tests",
                       "test_id in (%s)" % test_ids)
    label_map = {}
    for test_id, testlabel_id in labels:
        label_map.setdefault(test_id, []).append(testlabel_id)

    tests = []
    for raw_test in raw_tests:
        tests.append(models.test(raw_test[1], raw_test[2], None, None, None,
                                 None, raw_test[3], raw_test[4],
                                 [], {}, label_map.get(raw_test[0], [])))
    return tests


def parse_one(db, jobname, path, reparse, mail_on_failure):
    """
    Parse a single job. Optionally send email on failure.
    """
    tko_utils.dprint("\nScanning %s (%s)" % (jobname, path))
    old_job_idx = db.find_job(jobname)
    old_tests = []
    if reparse and old_job_idx:
        tko_utils.dprint("! Deleting old copy of job results to "
                         "reparse it")
        old_tests = find_old_tests(db, old_job_idx)
        db.delete_job(jobname)
    if db.find_job(jobname):
        tko_utils.dprint("! Job is already parsed, done")
        return

    # look up the status version
    job_keyval = models.job.read_keyval(path)
    status_version = job_keyval.get("status_version", 0)

    # parse out the job
    parser = status_lib.parser(status_version)
    job = parser.make_job(path)
    status_log = os.path.join(path, "status.log")
    if not os.path.exists(status_log):
        status_log = os.path.join(path, "status")
    if not os.path.exists(status_log):
        tko_utils.dprint("! Unable to parse job, no status file")
        return

    # parse the status logs
    tko_utils.dprint("+ Parsing dir=%s, jobname=%s" % (path, jobname))
    status_lines = open(status_log).readlines()
    parser.start(job)
    tests = parser.end(status_lines)

    # parser.end can return the same object multiple times, so filter out dups
    job.tests = []
    already_added = set()
    for test in tests:
        if test not in already_added:
            already_added.add(test)
            job.tests.append(test)

    # try and port labels over from the old tests, but if old tests stop
    # matching up with new ones just give up
    for test, old_test in zip(job.tests, old_tests):
        tests_are_the_same = (test.testname == old_test.testname and
                              test.subdir == old_test.subdir and
                              test.started_time == old_test.started_time and
                              (test.finished_time == old_test.finished_time or
                               old_test.finished_time is None))
        if tests_are_the_same:
            test.labels = old_test.labels
        else:
            tko_utils.dprint("! Reparse returned new tests, "
                             "dropping old test labels")

    # check for failures
    message_lines = [""]
    for test in job.tests:
        if not test.subdir:
            continue
        tko_utils.dprint("* testname, status, reason: %s %s %s"
                         % (test.subdir, test.status, test.reason))
        if test.status in ("FAIL", "WARN"):
            message_lines.append(format_failure_message(
                jobname, test.kernel.base, test.subdir,
                test.status, test.reason))
    message = "\n".join(message_lines)

    # send out a email report of failure
    if len(message) > 2 and mail_on_failure:
        tko_utils.dprint("Sending email report of failure on %s to %s"
                         % (jobname, job.user))
        mailfailure(jobname, job, message)

    # write the job into the database
    db.insert_job(jobname, job)
    db.commit()


def _get_job_subdirs(path):
    """
    Returns a list of job subdirectories at path. Returns None if the test
    is itself a job directory. Does not recurse into the subdirs.
    """
    # if there's a .machines file, use it to get the subdirs
    machine_list = os.path.join(path, ".machines")
    if os.path.exists(machine_list):
        subdirs = set(line.strip() for line in file(machine_list))
        existing_subdirs = set(subdir for subdir in subdirs
                               if os.path.exists(os.path.join(path, subdir)))
        if len(existing_subdirs) != 0:
            return existing_subdirs

    # if this dir contains ONLY subdirectories, return them
    contents = set(os.listdir(path))
    contents.discard(".parse.lock")
    subdirs = set(sub for sub in contents if
                  os.path.isdir(os.path.join(path, sub)))
    if len(contents) == len(subdirs) != 0:
        return subdirs

    # this is a job directory, or something else we don't understand
    return None


def parse_leaf_path(db, path, level, reparse, mail_on_failure):
    job_elements = path.split("/")[-level:]
    jobname = "/".join(job_elements)
    try:
        db.run_with_retry(parse_one, db, jobname, path, reparse,
                          mail_on_failure)
    except Exception:
        traceback.print_exc()


def parse_path(db, path, level, reparse, mail_on_failure):
    job_subdirs = _get_job_subdirs(path)
    if job_subdirs is not None:
        # parse status.log in current directory, if it exists. multi-machine
        # synchronous server side tests record output in this directory. without
        # this check, we do not parse these results.
        if os.path.exists(os.path.join(path, 'status.log')):
            parse_leaf_path(db, path, level, reparse, mail_on_failure)
        # multi-machine job
        for subdir in job_subdirs:
            jobpath = os.path.join(path, subdir)
            parse_path(db, jobpath, level + 1, reparse, mail_on_failure)
    else:
        # single machine job
        parse_leaf_path(db, path, level, reparse, mail_on_failure)


def _site_post_parse_job_dummy():
    return {}


def main():
    options, args = parse_args()
    results_dir = os.path.abspath(args[0])
    assert os.path.exists(results_dir)

    pid_file_manager = pidfile.PidFileManager("parser", results_dir)

    if options.write_pidfile:
        pid_file_manager.open_file()

    site_post_parse_job = utils.import_site_function(__file__,
        "autotest_lib.tko.site_parse", "site_post_parse_job",
        _site_post_parse_job_dummy)

    try:
        # build up the list of job dirs to parse
        if options.singledir:
            jobs_list = [results_dir]
        else:
            jobs_list = [os.path.join(results_dir, subdir)
                         for subdir in os.listdir(results_dir)]

        # build up the database
        db = tko_db.db(autocommit=False, host=options.db_host,
                       user=options.db_user, password=options.db_pass,
                       database=options.db_name)

        # parse all the jobs
        for path in jobs_list:
            lockfile = open(os.path.join(path, ".parse.lock"), "w")
            flags = fcntl.LOCK_EX
            if options.noblock:
                flags |= fcntl.LOCK_NB
            try:
                fcntl.flock(lockfile, flags)
            except IOError, e:
                # lock is not available and nonblock has been requested
                if e.errno == errno.EWOULDBLOCK:
                    lockfile.close()
                    continue
                else:
                    raise # something unexpected happened
            try:
                parse_path(db, path, options.level, options.reparse,
                           options.mailit)

            finally:
                fcntl.flock(lockfile, fcntl.LOCK_UN)
                lockfile.close()

        if options.site_do_post is True:
            site_post_parse_job(results_dir)

    except:
        pid_file_manager.close_file(1)
        raise
    else:
        pid_file_manager.close_file(0)


if __name__ == "__main__":
    main()