Source code for galaxy.jobs.runners.util.condor

"""
Condor helper utilities.
"""
from subprocess import (
    CalledProcessError,
    check_call,
)

from galaxy.util import (
    commands,
    unicodify,
)
from ..external import parse_external_id

DEFAULT_QUERY_CLASSAD = dict(
    universe="vanilla",
    getenv="true",
    notification="NEVER",
)

PROBLEM_PARSING_EXTERNAL_ID = "Failed to find job id from condor_submit"

SUBMIT_PARAM_PREFIX = "submit_"


[docs]def submission_params(prefix=SUBMIT_PARAM_PREFIX, **kwds):
    submission_params = {}
    prefix_len = len(prefix)
    for key in kwds:
        value = kwds[key]
        key = key.lower()
        if key.startswith(prefix):
            condor_key = key[prefix_len:]
            submission_params[condor_key] = value
    return submission_params


[docs]def build_submit_description(executable, output, error, user_log, query_params):
    """
    Build up the contents of a condor submit description file.

    >>> submit_args = dict(executable='/path/to/script', output='o', error='e', user_log='ul')
    >>> submit_args['query_params'] = dict()
    >>> default_description = build_submit_description(**submit_args)
    >>> assert 'executable = /path/to/script' in default_description
    >>> assert 'output = o' in default_description
    >>> assert 'error = e' in default_description
    >>> assert 'queue' in default_description
    >>> assert 'universe = vanilla' in default_description
    >>> assert 'universe = standard' not in default_description
    >>> submit_args['query_params'] = dict(universe='standard')
    >>> std_description = build_submit_description(**submit_args)
    >>> assert 'universe = vanilla' not in std_description
    >>> assert 'universe = standard' in std_description
    """
    all_query_params = DEFAULT_QUERY_CLASSAD.copy()
    all_query_params.update(query_params)

    submit_description = []
    for key, value in all_query_params.items():
        submit_description.append(f"{key} = {value}")
    submit_description.append(f"executable = {executable}")
    submit_description.append(f"output = {output}")
    submit_description.append(f"error = {error}")
    submit_description.append(f"log = {user_log}")
    submit_description.append("queue")
    return "\n".join(submit_description)


[docs]def condor_submit(submit_file):
    """
    Submit a condor job described by the given file. Parse an external id for
    the submission or return None and a reason for the failure.
    """
    external_id = None
    failure_message = None
    try:
        condor_message = commands.execute(("condor_submit", submit_file))
    except commands.CommandLineException as e:
        failure_message = unicodify(e)
    else:
        try:
            external_id = parse_external_id(condor_message, type="condor")
        except Exception:
            failure_message = f"{PROBLEM_PARSING_EXTERNAL_ID}: {condor_message}"
    return external_id, failure_message


[docs]def condor_stop(external_id):
    """
    Stop running condor job and return a failure_message if this
    fails.
    """
    failure_message = None
    try:
        check_call(("condor_rm", external_id))
    except CalledProcessError:
        failure_message = "condor_rm failed"
    except Exception as e:
        failure_message = f"error encountered calling condor_rm: {unicodify(e)}"
    return failure_message


[docs]def summarize_condor_log(log_file, external_id):
    """ """
    log_job_id = external_id.zfill(3)
    s1 = s4 = s7 = s5 = s9 = False
    with open(log_file) as log_handle:
        for line in log_handle:
            if f"001 ({log_job_id}." in line:
                s1 = True
            if f"004 ({log_job_id}." in line:
                s4 = True
            if f"007 ({log_job_id}." in line:
                s7 = True
            if f"005 ({log_job_id}." in line:
                s5 = True
            if f"009 ({log_job_id}." in line:
                s9 = True
        file_size = log_handle.tell()
    return s1, s4, s7, s5, s9, file_size