# Copyright 2015 The Chromium OS Authors. All rights reserved.
|
# Use of this source code is governed by a BSD-style license that can be
|
# found in the LICENSE file.
|
|
"""Sequence extensions to server_job.
|
Adds ability to schedule jobs on given machines.
|
"""
|
|
import logging
|
import os
|
|
import common
|
from autotest_lib.client.common_lib import control_data
|
from autotest_lib.client.common_lib import priorities
|
from autotest_lib.server import utils
|
from autotest_lib.server.cros.dynamic_suite import control_file_getter
|
from autotest_lib.server.cros.dynamic_suite import frontend_wrappers
|
from autotest_lib.site_utils import job_directories
|
|
MINUTE_IN_SECS = 60
|
HOUR_IN_MINUTES = 60
|
HOUR_IN_SECS = HOUR_IN_MINUTES * MINUTE_IN_SECS
|
DAY_IN_HOURS = 24
|
DAY_IN_SECS = DAY_IN_HOURS*HOUR_IN_SECS
|
|
DEFAULT_JOB_TIMEOUT_IN_MINS = 4 * HOUR_IN_MINUTES
|
|
class SequenceJob(object):
|
"""Define part of a sequence that will be scheduled by the sequence test."""
|
|
CONTROL_FILE = """
|
def run(machine):
|
job.run_test('%s', host=hosts.create_host(machine), client_ip=machine%s)
|
|
parallel_simple(run, machines)
|
"""
|
|
def __init__(self, name, args=None, iteration=1, duration=None,
|
fetch_control_file=False):
|
"""
|
Constructor
|
|
@param name: name of the server test to run.
|
@param args: arguments needed by the server test.
|
@param iteration: number of copy of this test to sechudle
|
@param duration: expected duration of the test (in seconds).
|
@param fetch_control_file: If True, fetch the control file contents
|
from disk. Otherwise uses the template
|
control file.
|
"""
|
self._name = name
|
self._args = args or {}
|
self._iteration = iteration
|
self._duration = duration
|
self._fetch_control_file = fetch_control_file
|
|
|
def child_job_name(self, machine, iteration_number):
|
"""
|
Return a name for a child job.
|
|
@param machine: machine name on which the test will run.
|
@param iteration_number: number with 0 and self._iteration - 1.
|
|
@returns a unique name based on the machine, the name and the iteration.
|
"""
|
name_parts = [machine, self._name]
|
tag = self._args.get('tag')
|
if tag:
|
name_parts.append(tag)
|
if self._iteration > 1:
|
name_parts.append(str(iteration_number))
|
return '_'.join(name_parts)
|
|
|
def child_job_timeout(self):
|
"""
|
Get the child job timeout in minutes.
|
|
@param args: arguments sent to the test.
|
|
@returns a timeout value for the test, 4h by default.
|
"""
|
if self._duration:
|
return 2 * int(self._duration) / MINUTE_IN_SECS
|
# default value:
|
return DEFAULT_JOB_TIMEOUT_IN_MINS
|
|
|
def child_control_file(self):
|
"""
|
Generate the child job's control file.
|
|
If not fetching the contents, use the template control file and
|
populate the template control file with the test name and expand the
|
arguments list.
|
|
@param test: name of the test to run
|
@param args: dictionary of argument for this test.
|
|
@returns a fully built control file to be use for the child job.
|
"""
|
if self._fetch_control_file:
|
# TODO (sbasi): Add arg support.
|
cntl_file_getter = control_file_getter.FileSystemGetter(
|
[os.path.join(os.path.dirname(os.path.realpath(__file__)),
|
'..')])
|
return cntl_file_getter.get_control_file_contents_by_name(
|
self._name)
|
child_args = ['',]
|
for arg, value in self._args.iteritems():
|
child_args.append('%s=%s' % (arg, repr(value)))
|
if self._duration:
|
child_args.append('duration=%d' % self._duration)
|
return self.CONTROL_FILE % (self._name, ', '.join(child_args))
|
|
|
def schedule(self, job, timeout_mins, machine):
|
"""
|
Sequence a job on the running AFE.
|
|
Will schedule a given test on the job machine(s).
|
Support a subset of tests:
|
- server job
|
- no hostless.
|
- no cleanup around tests.
|
|
@param job: server_job object that will server as parent.
|
@param timeout_mins: timeout to set up: if the test last more than
|
timeout_mins, the test will fail.
|
@param machine: machine to run the test on.
|
|
@returns a maximal time in minutes that the sequence can take.
|
"""
|
afe = frontend_wrappers.RetryingAFE(timeout_min=30, delay_sec=10,
|
user=job.user, debug=False)
|
# job_directores.get_job_id_or_task_id() will return a non-int opaque id
|
# for Chrome OS Skylab tasks. But sequences will break in that case
|
# anyway, because they try to create AFE jobs internally.
|
current_job_id = int(
|
job_directories.get_job_id_or_task_id(job.resultdir))
|
logging.debug('Current job id: %s', current_job_id)
|
runtime_mins = self.child_job_timeout()
|
hostname = utils.get_hostname_from_machine(machine)
|
|
for i in xrange(0, self._iteration):
|
child_job_name = self.child_job_name(hostname, i)
|
logging.debug('Creating job: %s', child_job_name)
|
afe.create_job(
|
self.child_control_file(),
|
name=child_job_name,
|
priority=priorities.Priority.DEFAULT,
|
control_type=control_data.CONTROL_TYPE.SERVER,
|
hosts=[hostname], meta_hosts=(), one_time_hosts=(),
|
synch_count=None, is_template=False,
|
timeout_mins=timeout_mins + (i + 1) * runtime_mins,
|
max_runtime_mins=runtime_mins,
|
run_verify=False, email_list='', dependencies=(),
|
reboot_before=None, reboot_after=None,
|
parse_failed_repair=None,
|
hostless=False, keyvals=None,
|
drone_set=None, image=None,
|
parent_job_id=current_job_id, test_retry=0, run_reset=False,
|
require_ssp=utils.is_in_container())
|
return runtime_mins * self._iteration
|
|
|
def sequence_schedule(job, machines, server_tests):
|
"""
|
Schedule the tests to run
|
|
Launch all the tests in the sequence on all machines.
|
Returns as soon as the jobs are launched.
|
|
@param job: Job running.
|
@param machines: machine to run on.
|
@param server_tests: Array of sequence_test objects.
|
"""
|
for machine in machines:
|
timeout_mins = 0
|
for test in server_tests:
|
timeout_mins += test.schedule(job, timeout_mins, machine)
|
