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 current_job_id = job_directories.get_job_id_or_task_id(job.resultdir) 138 logging.debug('Current job id: %s', current_job_id) 139 runtime_mins = self.child_job_timeout() 140 hostname = utils.get_hostname_from_machine(machine) 141 142 for i in xrange(0, self._iteration): 143 child_job_name = self.child_job_name(hostname, i) 144 logging.debug('Creating job: %s', child_job_name) 145 afe.create_job( 146 self.child_control_file(), 147 name=child_job_name, 148 priority=priorities.Priority.DEFAULT, 149 control_type=control_data.CONTROL_TYPE.SERVER, 150 hosts=[hostname], meta_hosts=(), one_time_hosts=(), 151 synch_count=None, is_template=False, 152 timeout_mins=timeout_mins + (i + 1) * runtime_mins, 153 max_runtime_mins=runtime_mins, 154 run_verify=False, email_list='', dependencies=(), 155 reboot_before=None, reboot_after=None, 156 parse_failed_repair=None, 157 hostless=False, keyvals=None, 158 drone_set=None, image=None, 159 parent_job_id=current_job_id, test_retry=0, run_reset=False, 160 require_ssp=utils.is_in_container()) 161 return runtime_mins * self._iteration 162 163 164 def sequence_schedule(job, machines, server_tests): 165 """ 166 Schedule the tests to run 167 168 Launch all the tests in the sequence on all machines. 169 Returns as soon as the jobs are launched. 170 171 @param job: Job running. 172 @param machines: machine to run on. 173 @param server_tests: Array of sequence_test objects. 174 """ 175 for machine in machines: 176 timeout_mins = 0 177 for test in server_tests: 178 timeout_mins += test.schedule(job, timeout_mins, machine) 179
