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