blob: ddf4e68fe15ac6e831f32803b8f461dd21b3b721 [file] [log] [blame]
# Copyright 2008 Google Inc. All Rights Reserved.
The job module contains the objects and methods used to
manage jobs in Autotest.
The valid actions are:
list: lists job(s)
create: create a job
abort: abort job(s)
stat: detailed listing of job(s)
The common options are:
See for a High Level Design and Algorithm.
import getpass, os, pwd, re, socket, sys
from autotest_lib.cli import topic_common, action_common
from autotest_lib.client.common_lib import control_data
class job(topic_common.atest):
"""Job class
atest job [create|clone|list|stat|abort] <options>"""
usage_action = '[create|clone|list|stat|abort]'
topic = msg_topic = 'job'
msg_items = '<job_ids>'
def _convert_status(self, results):
for result in results:
total = sum(result['status_counts'].values())
status = ['%s=%s(%.1f%%)' % (key, val, 100.0*float(val)/total)
for key, val in result['status_counts'].iteritems()]
result['status_counts'] = ', '.join(status)
def backward_compatibility(self, action, argv):
""" 'job create --clone' became 'job clone --id' """
if action == 'create':
for option in ['-l', '--clone']:
if option in argv:
argv[argv.index(option)] = '--id'
action = 'clone'
return action
class job_help(job):
"""Just here to get the atest logic working.
Usage is set by its parent"""
class job_list_stat(action_common.atest_list, job):
def __init__(self):
super(job_list_stat, self).__init__()
self.topic_parse_info = topic_common.item_parse_info(
def __split_jobs_between_ids_names(self):
job_ids = []
job_names = []
# Sort between job IDs and names
for job_id in
if job_id.isdigit():
return (job_ids, job_names)
def execute_on_ids_and_names(self, op, filters={},
check_results={'id__in': 'id',
'name__in': 'id'},
tag_id='id__in', tag_name='name__in'):
if not
# Want everything
return super(job_list_stat, self).execute(op=op, filters=filters)
all_jobs = []
(job_ids, job_names) = self.__split_jobs_between_ids_names()
for items, tag in [(job_ids, tag_id),
(job_names, tag_name)]:
if items:
new_filters = filters.copy()
new_filters[tag] = items
jobs = super(job_list_stat,
return all_jobs
class job_list(job_list_stat):
"""atest job list [<jobs>] [--all] [--running] [--user <username>]"""
def __init__(self):
super(job_list, self).__init__()
self.parser.add_option('-a', '--all', help='List jobs for all '
'users.', action='store_true', default=False)
self.parser.add_option('-r', '--running', help='List only running '
'jobs', action='store_true')
self.parser.add_option('-u', '--user', help='List jobs for given '
'user', type='string')
def parse(self):
options, leftover = super(job_list, self).parse()
self.all = options.all['running'] = options.running
if options.user:
if options.all:
self.invalid_syntax('Only specify --all or --user, not both.')
else:['owner'] = options.user
elif not options.all and not['owner'] = getpass.getuser()
return options, leftover
def execute(self):
return self.execute_on_ids_and_names(op='get_jobs_summary',
def output(self, results):
keys = ['id', 'owner', 'name', 'status_counts']
if self.verbose:
keys.extend(['priority', 'control_type', 'created_on'])
super(job_list, self).output(results, keys)
class job_stat(job_list_stat):
"""atest job stat <job>"""
usage_action = 'stat'
def __init__(self):
super(job_stat, self).__init__()
self.parser.add_option('-f', '--control-file',
help='Display the control file',
action='store_true', default=False)
self.parser.add_option('-N', '--list-hosts',
help='Display only a list of hosts',
self.parser.add_option('-s', '--list-hosts-status',
help='Display only the hosts in these statuses '
'for a job.', action='store')
def parse(self):
status_list = topic_common.item_parse_info(
options, leftover = super(job_stat, self).parse([status_list],
if not
self.invalid_syntax('Must specify at least one job.')
self.show_control_file = options.control_file
self.list_hosts = options.list_hosts
if self.list_hosts and self.status_list:
self.invalid_syntax('--list-hosts is implicit when using '
if len( > 1 and (self.list_hosts or self.status_list):
self.invalid_syntax('--list-hosts and --list-hosts-status should '
'only be used on a single job.')
return options, leftover
def _merge_results(self, summary, qes):
hosts_status = {}
for qe in qes:
if qe['host']:
job_id = qe['job']['id']
hostname = qe['host']['hostname']
for job in summary:
job_id = job['id']
if hosts_status.has_key(job_id):
this_job = hosts_status[job_id]
job['hosts'] = ' '.join(' '.join(host) for host in
host_per_status = ['%s="%s"' %(status, ' '.join(host))
for status, host in this_job.iteritems()]
job['hosts_status'] = ', '.join(host_per_status)
if self.status_list:
statuses = set(s.lower() for s in self.status_list)
all_hosts = [s for s in host_per_status if s.split('=',
1)[0].lower() in statuses]
job['hosts_selected_status'] = '\n'.join(all_hosts)
job['hosts_status'] = ''
if not job.get('hosts'):
self.generic_error('Job has unassigned meta-hosts, '
'try again shortly.')
return summary
def execute(self):
summary = self.execute_on_ids_and_names(op='get_jobs_summary')
# Get the real hostnames
qes = self.execute_on_ids_and_names(op='get_host_queue_entries',
return self._merge_results(summary, qes)
def output(self, results):
if self.list_hosts:
keys = ['hosts']
elif self.status_list:
keys = ['hosts_selected_status']
elif not self.verbose:
keys = ['id', 'name', 'priority', 'status_counts', 'hosts_status']
keys = ['id', 'name', 'priority', 'status_counts', 'hosts_status',
'owner', 'control_type', 'synch_count', 'created_on',
'run_verify', 'reboot_before', 'reboot_after',
if self.show_control_file:
super(job_stat, self).output(results, keys)
class job_create_or_clone(action_common.atest_create, job):
"""Class containing the code common to the job create and clone actions"""
msg_items = 'job_name'
def __init__(self):
super(job_create_or_clone, self).__init__()
self.hosts = []
self.data_item_key = 'name'
self.parser.add_option('-p', '--priority', help='Job priority (low, '
'medium, high, urgent), default=medium',
type='choice', choices=('low', 'medium', 'high',
'urgent'), default='medium')
self.parser.add_option('-b', '--labels',
help='Comma separated list of labels '
'to get machine list from.', default='')
self.parser.add_option('-m', '--machine', help='List of machines to '
'run on')
self.parser.add_option('-M', '--mlist',
help='File listing machines to use',
type='string', metavar='MACHINE_FLIST')
help='List of one time hosts')
self.parser.add_option('-e', '--email',
help='A comma seperated list of '
'email addresses to notify of job completion',
def _parse_hosts(self, args):
""" Parses the arguments to generate a list of hosts and meta_hosts
A host is a regular name, a meta_host is n*label or *label.
These can be mixed on the CLI, and separated by either commas or
spaces, e.g.: 5*Machine_Label host0 5*Machine_Label2,host2 """
hosts = []
meta_hosts = []
for arg in args:
for host in arg.split(','):
if re.match('^[0-9]+[*]', host):
num, host = host.split('*', 1)
meta_hosts += int(num) * [host]
elif re.match('^[*](\w*)', host):
meta_hosts += [re.match('^[*](\w*)', host).group(1)]
elif host != '' and host not in hosts:
# Real hostname and not a duplicate
return (hosts, meta_hosts)
def parse(self, parse_info=[]):
host_info = topic_common.item_parse_info(attribute_name='hosts',
job_info = topic_common.item_parse_info(attribute_name='jobname',
oth_info = topic_common.item_parse_info(attribute_name='one_time_hosts',
label_info = topic_common.item_parse_info(attribute_name='labels',
options, leftover = super(job_create_or_clone, self).parse(
[host_info, job_info, oth_info, label_info] + parse_info,
req_items='jobname') = {}
jobname = getattr(self, 'jobname')
if len(jobname) > 1:
self.invalid_syntax('Too many arguments specified, only expected '
'to receive job name: %s' % jobname)
self.jobname = jobname[0]
if options.priority:['priority'] = options.priority.capitalize()
if self.one_time_hosts:['one_time_hosts'] = self.one_time_hosts
if self.labels:
label_hosts = self.execute_rpc(op='get_hosts',
for host in label_hosts:
self.hosts.append(host['hostname'])['name'] = self.jobname
(['hosts'],['meta_hosts']) = self._parse_hosts(self.hosts)['email_list'] =
return options, leftover
def create_job(self):
job_id = self.execute_rpc(op='create_job', **
return ['%s (id %s)' % (self.jobname, job_id)]
def get_items(self):
return [self.jobname]
class job_create(job_create_or_clone):
"""atest job create [--priority <Low|Medium|High|Urgent>]
[--synch_count] [--control-file </path/to/cfile>]
[--on-server] [--test <test1,test2>] [--kernel <http://kernel>]
[--mlist </path/to/machinelist>] [--machine <host1 host2 host3>]
[--labels <list of labels of machines to run on>]
[--reboot_before <option>] [--reboot_after <option>]
[--noverify] [--timeout <timeout>] [--max_runtime <max runtime>]
[--one-time-hosts <hosts>] [--email <email>]
[--dependencies <labels this job is dependent on>]
[--atomic_group <atomic group name>] [--parse-failed-repair <option>]
[--image <http://path/to/image>]
Creating a job is rather different from the other create operations,
so it only uses the __init__() and output() from its superclass.
op_action = 'create'
def __init__(self):
super(job_create, self).__init__()
self.ctrl_file_data = {}
self.parser.add_option('-y', '--synch_count', type=int,
help='Number of machines to use per autoserv '
self.parser.add_option('-f', '--control-file',
help='use this control file', metavar='FILE')
self.parser.add_option('-s', '--server',
help='This is server-side job',
action='store_true', default=False)
self.parser.add_option('-t', '--test',
help='List of tests to run')
self.parser.add_option('-k', '--kernel', help='A comma separated list'
' of kernel versions/URLs/filenames to run the'
' job on')
self.parser.add_option('--kernel-cmdline', help='A string that will be'
' given as cmdline to the booted kernel(s)'
' specified by the -k option')
self.parser.add_option('-d', '--dependencies', help='Comma separated '
'list of labels this job is dependent on.',
self.parser.add_option('-G', '--atomic_group', help='Name of an Atomic '
'Group to schedule this job on.',
self.parser.add_option('-B', '--reboot_before',
help='Whether or not to reboot the machine '
'before the job (never/if dirty/always)',
choices=('never', 'if dirty', 'always'))
self.parser.add_option('-a', '--reboot_after',
help='Whether or not to reboot the machine '
'after the job (never/if all tests passed/'
choices=('never', 'if all tests passed',
help='Whether or not to parse failed repair '
'results as part of the job',
choices=('true', 'false'))
self.parser.add_option('-n', '--noverify',
help='Do not run verify for job',
default=False, action='store_true')
self.parser.add_option('-o', '--timeout_mins',
help='Job timeout in minutes.',
help='Job maximum runtime in minutes')
self.parser.add_option('-i', '--image',
help='OS image to install before running the '
def _get_kernel_data(kernel_list, cmdline):
# the RPC supports cmdline per kernel version in a dictionary
kernels = []
for version in re.split(r'[, ]+', kernel_list):
if not version:
kernel_info = {'version': version}
if cmdline:
kernel_info['cmdline'] = cmdline
return kernels
def parse(self):
deps_info = topic_common.item_parse_info(attribute_name='dependencies',
options, leftover = super(job_create, self).parse(
if (len(self.hosts) == 0 and not self.one_time_hosts
and not options.labels and not options.atomic_group):
self.invalid_syntax('Must specify at least one machine '
'or an atomic group '
'(-m, -M, -b, -G or --one-time-hosts).')
if not options.control_file and not options.test:
self.invalid_syntax('Must specify either --test or --control-file'
' to create a job.')
if options.control_file and options.test:
self.invalid_syntax('Can only specify one of --control-file or '
'--test, not both.')
if options.kernel:
self.ctrl_file_data['kernel'] = self._get_kernel_data(
options.kernel, options.kernel_cmdline)
if options.control_file:
control_file_f = open(options.control_file)
control_file_data =
except IOError:
self.generic_error('Unable to read from specified '
'control-file: %s' % options.control_file)
if options.kernel:
# execute() will pass this to the AFE server to wrap this
# control file up to include the kernel installation steps.
self.ctrl_file_data['client_control_file'] = control_file_data
else:['control_file'] = control_file_data
if options.test:
if options.server:
self.invalid_syntax('If you specify tests, then the '
'client/server setting is implicit and '
'cannot be overriden.')
tests = [t.strip() for t in options.test.split(',') if t.strip()]
self.ctrl_file_data['tests'] = tests
if options.image:['image'] = options.image
if options.reboot_before:['reboot_before'] = options.reboot_before.capitalize()
if options.reboot_after:['reboot_after'] = options.reboot_after.capitalize()
if options.parse_failed_repair:['parse_failed_repair'] = (
options.parse_failed_repair == 'true')
if options.noverify:['run_verify'] = False
if options.timeout_mins:['timeout_mins'] = options.timeout_mins
if options.max_runtime:['max_runtime_mins'] = options.max_runtime
if options.atomic_group:['atomic_group_name'] = options.atomic_group['dependencies'] = self.dependencies
if options.synch_count:['synch_count'] = options.synch_count
if options.server:['control_type'] = control_data.CONTROL_TYPE_NAMES.SERVER
else:['control_type'] = control_data.CONTROL_TYPE_NAMES.CLIENT
return options, leftover
def execute(self):
if self.ctrl_file_data:
uploading_kernel = 'kernel' in self.ctrl_file_data
if uploading_kernel:
default_timeout = socket.getdefaulttimeout()
print 'Uploading Kernel: this may take a while...',
cf_info = self.execute_rpc(op='generate_control_file',
if uploading_kernel:
if uploading_kernel:
print 'Done'['control_file'] = cf_info['control_file']
if 'synch_count' not in['synch_count'] = cf_info['synch_count']
if cf_info['is_server']:['control_type'] = control_data.CONTROL_TYPE_NAMES.SERVER
else:['control_type'] = control_data.CONTROL_TYPE_NAMES.CLIENT
# Get the union of the 2 sets of dependencies
deps = set(['dependencies'])
deps = sorted(deps.union(cf_info['dependencies']))['dependencies'] = list(deps)
if 'synch_count' not in['synch_count'] = 1
return self.create_job()
class job_clone(job_create_or_clone):
"""atest job clone [--priority <Low|Medium|High|Urgent>]
[--mlist </path/to/machinelist>] [--machine <host1 host2 host3>]
[--labels <list of labels of machines to run on>]
[--one-time-hosts <hosts>] [--email <email>]
Cloning a job is rather different from the other create operations,
so it only uses the __init__() and output() from its superclass.
op_action = 'clone'
usage_action = 'clone'
def __init__(self):
super(job_clone, self).__init__()
self.parser.add_option('-i', '--id', help='Job id to clone',
self.parser.add_option('-r', '--reuse-hosts',
help='Use the exact same hosts as the '
'cloned job.',
action='store_true', default=False)
def parse(self):
options, leftover = super(job_clone, self).parse()
self.clone_id =
self.reuse_hosts = options.reuse_hosts
host_specified = self.hosts or self.one_time_hosts or options.labels
if self.reuse_hosts and host_specified:
self.invalid_syntax('Cannot specify hosts and reuse the same '
'ones as the cloned job.')
if not (self.reuse_hosts or host_specified):
self.invalid_syntax('Must reuse or specify at least one '
'machine (-r, -m, -M, -b or '
return options, leftover
def execute(self):
clone_info = self.execute_rpc(op='get_info_for_clone',
# Remove fields from clone data that cannot be reused
for field in ('name', 'created_on', 'id', 'owner'):
del clone_info['job'][field]
# Also remove parameterized_job field, as the feature still is
# incomplete, this tool does not attempt to support it for now,
# it uses a different API function and it breaks create_job()
if clone_info['job'].has_key('parameterized_job'):
del clone_info['job']['parameterized_job']
# Keyword args cannot be unicode strings, val)
for key, val in clone_info['job'].iteritems())
if self.reuse_hosts:
# Convert host list from clone info that can be used for job_create
for label, qty in clone_info['meta_host_counts'].iteritems():['meta_hosts'].extend([label]*qty)['hosts'].extend(host['hostname']
for host in clone_info['hosts'])
return self.create_job()
class job_abort(job, action_common.atest_delete):
"""atest job abort <job(s)>"""
usage_action = op_action = 'abort'
msg_done = 'Aborted'
def parse(self):
job_info = topic_common.item_parse_info(attribute_name='jobids',
options, leftover = super(job_abort, self).parse([job_info],
def execute(self):
data = {'job__id__in': self.jobids}
self.execute_rpc(op='abort_host_queue_entries', **data)
print 'Aborting jobs: %s' % ', '.join(self.jobids)
def get_items(self):
return self.jobids