1 # Copyright 2015 The Chromium OS Authors. All rights reserved. 2 # Use of this source code is governed by a BSD-style license that can be 3 # found in the LICENSE file. 4 5 """Sequence extensions to server_job. 6 Adds ability to schedule jobs on given machines. 7 """ 8 9 import logging 10 import os 11 12 import common 13 from autotest_lib.client.common_lib import control_data 14 from autotest_lib.client.common_lib import priorities 15 from autotest_lib.server import utils 16 from autotest_lib.server.cros.dynamic_suite import control_file_getter 17 from autotest_lib.server.cros.dynamic_suite import frontend_wrappers 18 from autotest_lib.site_utils import job_directories 19 20 MINUTE_IN_SECS = 60 21 HOUR_IN_MINUTES = 60 22 HOUR_IN_SECS = HOUR_IN_MINUTES * MINUTE_IN_SECS 23 DAY_IN_HOURS = 24 24 DAY_IN_SECS = DAY_IN_HOURS*HOUR_IN_SECS 25 26 DEFAULT_JOB_TIMEOUT_IN_MINS = 4 * HOUR_IN_MINUTES 27 28 class SequenceJob(object): 29 """Define part of a sequence that will be scheduled by the sequence test.""" 30 31 CONTROL_FILE = """ 32 def run(machine): 33 job.run_test('%s', host=hosts.create_host(machine), client_ip=machine%s) 34 35 parallel_simple(run, machines) 36 """ 37 38 def __init__(self, name, args=None, iteration=1, duration=None, 39 fetch_control_file=False): 40 """ 41 Constructor 42 43 @param name: name of the server test to run. 44 @param args: arguments needed by the server test. 45 @param iteration: number of copy of this test to sechudle 46 @param duration: expected duration of the test (in seconds). 47 @param fetch_control_file: If True, fetch the control file contents 48 from disk. Otherwise uses the template 49 control file. 50 """ 51 self._name = name 52 self._args = args or {} 53 self._iteration = iteration 54 self._duration = duration 55 self._fetch_control_file = fetch_control_file 56 57 58 def child_job_name(self, machine, iteration_number): 59 """ 60 Return a name for a child job. 61 62 @param machine: machine name on which the test will run. 63 @param iteration_number: number with 0 and self._iteration - 1. 64 65 @returns a unique name based on the machine, the name and the iteration. 66 """ 67 name_parts = [machine, self._name] 68 tag = self._args.get('tag') 69 if tag: 70 name_parts.append(tag) 71 if self._iteration > 1: 72 name_parts.append(str(iteration_number)) 73 return '_'.join(name_parts) 74 75 76 def child_job_timeout(self): 77 """ 78 Get the child job timeout in minutes. 79 80 @param args: arguments sent to the test. 81 82 @returns a timeout value for the test, 4h by default. 83 """ 84 if self._duration: 85 return 2 * int(self._duration) / MINUTE_IN_SECS 86 # default value: 87 return DEFAULT_JOB_TIMEOUT_IN_MINS 88 89 90 def child_control_file(self): 91 """ 92 Generate the child job's control file. 93 94 If not fetching the contents, use the template control file and 95 populate the template control file with the test name and expand the 96 arguments list. 97 98 @param test: name of the test to run 99 @param args: dictionary of argument for this test. 100 101 @returns a fully built control file to be use for the child job. 102 """ 103 if self._fetch_control_file: 104 # TODO (sbasi): Add arg support. 105 cntl_file_getter = control_file_getter.FileSystemGetter( 106 [os.path.join(os.path.dirname(os.path.realpath(__file__)), 107 '..')]) 108 return cntl_file_getter.get_control_file_contents_by_name( 109 self._name) 110 child_args = ['',] 111 for arg, value in self._args.iteritems(): 112 child_args.append('%s=%s' % (arg, repr(value))) 113 if self._duration: 114 child_args.append('duration=%d' % self._duration) 115 return self.CONTROL_FILE % (self._name, ', '.join(child_args)) 116 117 118 def schedule(self, job, timeout_mins, machine): 119 """ 120 Sequence a job on the running AFE. 121 122 Will schedule a given test on the job machine(s). 123 Support a subset of tests: 124 - server job 125 - no hostless. 126 - no cleanup around tests. 127 128 @param job: server_job object that will server as parent. 129 @param timeout_mins: timeout to set up: if the test last more than 130 timeout_mins, the test will fail. 131 @param machine: machine to run the test on. 132 133 @returns a maximal time in minutes that the sequence can take. 134 """ 135 afe = frontend_wrappers.RetryingAFE(timeout_min=30, delay_sec=10, 136 user=job.user, debug=False) 137 # job_directores.get_job_id_or_task_id() will return a non-int opaque id 138 # for Chrome OS Skylab tasks. But sequences will break in that case 139 # anyway, because they try to create AFE jobs internally. 140 current_job_id = int( 141 job_directories.get_job_id_or_task_id(job.resultdir)) 142 logging.debug('Current job id: %s', current_job_id) 143 runtime_mins = self.child_job_timeout() 144 hostname = utils.get_hostname_from_machine(machine) 145 146 for i in xrange(0, self._iteration): 147 child_job_name = self.child_job_name(hostname, i) 148 logging.debug('Creating job: %s', child_job_name) 149 afe.create_job( 150 self.child_control_file(), 151 name=child_job_name, 152 priority=priorities.Priority.DEFAULT, 153 control_type=control_data.CONTROL_TYPE.SERVER, 154 hosts=[hostname], meta_hosts=(), one_time_hosts=(), 155 synch_count=None, is_template=False, 156 timeout_mins=timeout_mins + (i + 1) * runtime_mins, 157 max_runtime_mins=runtime_mins, 158 run_verify=False, email_list='', dependencies=(), 159 reboot_before=None, reboot_after=None, 160 parse_failed_repair=None, 161 hostless=False, keyvals=None, 162 drone_set=None, image=None, 163 parent_job_id=current_job_id, test_retry=0, run_reset=False, 164 require_ssp=utils.is_in_container()) 165 return runtime_mins * self._iteration 166 167 168 def sequence_schedule(job, machines, server_tests): 169 """ 170 Schedule the tests to run 171 172 Launch all the tests in the sequence on all machines. 173 Returns as soon as the jobs are launched. 174 175 @param job: Job running. 176 @param machines: machine to run on. 177 @param server_tests: Array of sequence_test objects. 178 """ 179 for machine in machines: 180 timeout_mins = 0 181 for test in server_tests: 182 timeout_mins += test.schedule(job, timeout_mins, machine) 183
