| #!/usr/bin/python2 |
| #pylint: disable-msg=C0111 |
| |
| # Copyright (c) 2014 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. |
| |
| """Host scheduler. |
| |
| If run as a standalone service, the host scheduler ensures the following: |
| 1. Hosts will not be assigned to multiple hqes simultaneously. The process |
| of assignment in this case refers to the modification of the host_id |
| column of a row in the afe_host_queue_entries table, to reflect the host |
| id of a leased host matching the dependencies of the job. |
| 2. Hosts that are not being used by active hqes or incomplete special tasks |
| will be released back to the available hosts pool, for acquisition by |
| subsequent hqes. |
| In addition to these guarantees, the host scheduler also confirms that no 2 |
| active hqes/special tasks are assigned the same host, and sets the leased bit |
| for hosts needed by frontend special tasks. The need for the latter is only |
| apparent when viewed in the context of the job-scheduler (monitor_db), which |
| runs special tasks only after their hosts have been leased. |
| |
| ** Suport minimum duts requirement for suites (non-inline mode) ** |
| |
| Each suite can specify the minimum number of duts it requires by |
| dropping a 'suite_min_duts' job keyval which defaults to 0. |
| |
| When suites are competing for duts, if any suite has not got minimum duts |
| it requires, the host scheduler will try to meet the requirement first, |
| even if other suite may have higher priority or earlier timestamp. Once |
| all suites' minimum duts requirement have been fullfilled, the host |
| scheduler will allocate the rest of duts based on job priority and suite job id. |
| This is to prevent low priority suites from starving when sharing pool with |
| high-priority suites. |
| |
| Note: |
| 1. Prevent potential starvation: |
| We need to carefully choose |suite_min_duts| for both low and high |
| priority suites. If a high priority suite didn't specify it but a low |
| priority one does, the high priority suite can be starved! |
| 2. Restart requirement: |
| Restart host scheduler if you manually released a host by setting |
| leased=0 in db. This is needed because host scheduler maintains internal |
| state of host assignment for suites. |
| 3. Exchanging duts triggers provisioning: |
| TODO(fdeng): There is a chance two suites can exchange duts, |
| if the two suites are for different builds, the exchange |
| will trigger provisioning. This can be optimized by preferring getting |
| hosts with the same build. |
| """ |
| |
| import argparse |
| import collections |
| import datetime |
| import logging |
| import os |
| import signal |
| import sys |
| import time |
| |
| import common |
| from autotest_lib.client.common_lib import utils |
| from autotest_lib.frontend import setup_django_environment |
| |
| # This import needs to come earlier to avoid using autotest's version of |
| # httplib2, which is out of date. |
| try: |
| from chromite.lib import metrics |
| from chromite.lib import ts_mon_config |
| except ImportError: |
| metrics = utils.metrics_mock |
| ts_mon_config = utils.metrics_mock |
| |
| from autotest_lib.client.common_lib import global_config |
| from autotest_lib.scheduler import email_manager |
| from autotest_lib.scheduler import query_managers |
| from autotest_lib.scheduler import rdb_lib |
| from autotest_lib.scheduler import rdb_utils |
| from autotest_lib.scheduler import scheduler_lib |
| from autotest_lib.scheduler import scheduler_models |
| from autotest_lib.site_utils import server_manager_utils |
| |
| |
| _db_manager = None |
| _shutdown = False |
| _tick_pause_sec = global_config.global_config.get_config_value( |
| 'SCHEDULER', 'tick_pause_sec', type=int, default=5) |
| _monitor_db_host_acquisition = global_config.global_config.get_config_value( |
| 'SCHEDULER', 'inline_host_acquisition', type=bool, default=True) |
| _METRICS_PREFIX = 'chromeos/autotest/host_scheduler' |
| |
| class SuiteRecorder(object): |
| """Recording the host assignment for suites. |
| |
| The recorder holds two things: |
| * suite_host_num, records how many duts a suite is holding, |
| which is a map <suite_job_id -> num_of_hosts> |
| * hosts_to_suites, records which host is assigned to which |
| suite, it is a map <host_id -> suite_job_id> |
| The two datastructure got updated when a host is assigned to or released |
| by a job. |
| |
| The reason to maintain hosts_to_suites is that, when a host is released, |
| we need to know which suite it was leased to. Querying the db for the |
| latest completed job that has run on a host is slow. Therefore, we go with |
| an alternative: keeping a <host id, suite job id> map |
| in memory (for 10K hosts, the map should take less than 1M memory on |
| 64-bit machine with python 2.7) |
| |
| """ |
| |
| |
| def __init__(self, job_query_manager): |
| """Initialize. |
| |
| @param job_queue_manager: A JobQueueryManager object. |
| """ |
| self.job_query_manager = job_query_manager |
| self.suite_host_num, self.hosts_to_suites = ( |
| self.job_query_manager.get_suite_host_assignment()) |
| |
| |
| def record_assignment(self, queue_entry): |
| """Record that the hqe has got a host. |
| |
| @param queue_entry: A scheduler_models.HostQueueEntry object which has |
| got a host. |
| """ |
| parent_id = queue_entry.job.parent_job_id |
| if not parent_id: |
| return |
| if self.hosts_to_suites.get(queue_entry.host_id, None) == parent_id: |
| logging.error('HQE (id: %d, parent_job_id: %d, host: %s) ' |
| 'seems already recorded', queue_entry.id, |
| parent_id, queue_entry.host.hostname) |
| return |
| num_hosts = self.suite_host_num.get(parent_id, 0) |
| self.suite_host_num[parent_id] = num_hosts + 1 |
| self.hosts_to_suites[queue_entry.host_id] = parent_id |
| logging.debug('Suite %d got host %s, currently holding %d hosts', |
| parent_id, queue_entry.host.hostname, |
| self.suite_host_num[parent_id]) |
| |
| |
| def record_release(self, hosts): |
| """Update the record with host releasing event. |
| |
| @param hosts: A list of scheduler_models.Host objects. |
| """ |
| for host in hosts: |
| if host.id in self.hosts_to_suites: |
| parent_job_id = self.hosts_to_suites.pop(host.id) |
| count = self.suite_host_num[parent_job_id] - 1 |
| if count == 0: |
| del self.suite_host_num[parent_job_id] |
| else: |
| self.suite_host_num[parent_job_id] = count |
| logging.debug( |
| 'Suite %d releases host %s, currently holding %d hosts', |
| parent_job_id, host.hostname, count) |
| |
| |
| def get_min_duts(self, suite_job_ids): |
| """Figure out min duts to request. |
| |
| Given a set ids of suite jobs, figure out minimum duts to request for |
| each suite. It is determined by two factors: min_duts specified |
| for each suite in its job keyvals, and how many duts a suite is |
| currently holding. |
| |
| @param suite_job_ids: A set of suite job ids. |
| |
| @returns: A dictionary, the key is suite_job_id, the value |
| is the minimum number of duts to request. |
| """ |
| suite_min_duts = self.job_query_manager.get_min_duts_of_suites( |
| suite_job_ids) |
| for parent_id in suite_job_ids: |
| min_duts = suite_min_duts.get(parent_id, 0) |
| cur_duts = self.suite_host_num.get(parent_id, 0) |
| suite_min_duts[parent_id] = max(0, min_duts - cur_duts) |
| logging.debug('Minimum duts to get for suites (suite_id: min_duts): %s', |
| suite_min_duts) |
| return suite_min_duts |
| |
| |
| class BaseHostScheduler(object): |
| """Base class containing host acquisition logic. |
| |
| This class contains all the core host acquisition logic needed by the |
| scheduler to run jobs on hosts. It is only capable of releasing hosts |
| back to the rdb through its tick, any other action must be instigated by |
| the job scheduler. |
| """ |
| |
| |
| host_assignment = collections.namedtuple('host_assignment', ['host', 'job']) |
| |
| |
| def __init__(self): |
| self.host_query_manager = query_managers.AFEHostQueryManager() |
| |
| |
| def _release_hosts(self): |
| """Release hosts to the RDB. |
| |
| Release all hosts that are ready and are currently not being used by an |
| active hqe, and don't have a new special task scheduled against them. |
| |
| @return a list of hosts that are released. |
| """ |
| release_hosts = self.host_query_manager.find_unused_healty_hosts() |
| release_hostnames = [host.hostname for host in release_hosts] |
| if release_hostnames: |
| self.host_query_manager.set_leased( |
| False, hostname__in=release_hostnames) |
| return release_hosts |
| |
| |
| @classmethod |
| def schedule_host_job(cls, host, queue_entry): |
| """Schedule a job on a host. |
| |
| Scheduling a job involves: |
| 1. Setting the active bit on the queue_entry. |
| 2. Scheduling a special task on behalf of the queue_entry. |
| Performing these actions will lead the job scheduler through a chain of |
| events, culminating in running the test and collecting results from |
| the host. |
| |
| @param host: The host against which to schedule the job. |
| @param queue_entry: The queue_entry to schedule. |
| """ |
| if queue_entry.host_id is None: |
| queue_entry.set_host(host) |
| elif host.id != queue_entry.host_id: |
| raise rdb_utils.RDBException('The rdb returned host: %s ' |
| 'but the job:%s was already assigned a host: %s. ' % |
| (host.hostname, queue_entry.job_id, |
| queue_entry.host.hostname)) |
| queue_entry.update_field('active', True) |
| |
| # TODO: crbug.com/373936. The host scheduler should only be assigning |
| # jobs to hosts, but the criterion we use to release hosts depends |
| # on it not being used by an active hqe. Since we're activating the |
| # hqe here, we also need to schedule its first prejob task. OTOH, |
| # we could converge to having the host scheduler manager all special |
| # tasks, since their only use today is to verify/cleanup/reset a host. |
| logging.info('Scheduling pre job tasks for entry: %s', queue_entry) |
| queue_entry.schedule_pre_job_tasks() |
| |
| |
| def acquire_hosts(self, host_jobs): |
| """Accquire hosts for given jobs. |
| |
| This method sends jobs that need hosts to rdb. |
| Child class can override this method to pipe more args |
| to rdb. |
| |
| @param host_jobs: A list of queue entries that either require hosts, |
| or require host assignment validation through the rdb. |
| |
| @param return: A generator that yields an rdb_hosts.RDBClientHostWrapper |
| for each host acquired on behalf of a queue_entry, |
| or None if a host wasn't found. |
| """ |
| return rdb_lib.acquire_hosts(host_jobs) |
| |
| |
| def find_hosts_for_jobs(self, host_jobs): |
| """Find and verify hosts for a list of jobs. |
| |
| @param host_jobs: A list of queue entries that either require hosts, |
| or require host assignment validation through the rdb. |
| @return: A generator of tuples of the form (host, queue_entry) for each |
| valid host-queue_entry assignment. |
| """ |
| hosts = self.acquire_hosts(host_jobs) |
| for host, job in zip(hosts, host_jobs): |
| if host: |
| yield self.host_assignment(host, job) |
| |
| |
| def tick(self): |
| """Schedule core host management activities.""" |
| self._release_hosts() |
| |
| |
| class HostScheduler(BaseHostScheduler): |
| """A scheduler capable managing host acquisition for new jobs.""" |
| |
| |
| def __init__(self): |
| super(HostScheduler, self).__init__() |
| self.job_query_manager = query_managers.AFEJobQueryManager() |
| # Keeping track on how many hosts each suite is holding |
| # {suite_job_id: num_hosts} |
| self._suite_recorder = SuiteRecorder(self.job_query_manager) |
| |
| |
| def _record_host_assignment(self, host, queue_entry): |
| """Record that |host| is assigned to |queue_entry|. |
| |
| Record: |
| 1. How long it takes to assign a host to a job in metadata db. |
| 2. Record host assignment of a suite. |
| |
| @param host: A Host object. |
| @param queue_entry: A HostQueueEntry object. |
| """ |
| secs_in_queued = (datetime.datetime.now() - |
| queue_entry.job.created_on).total_seconds() |
| self._suite_recorder.record_assignment(queue_entry) |
| |
| |
| @metrics.SecondsTimerDecorator( |
| '%s/schedule_jobs_duration' % _METRICS_PREFIX) |
| def _schedule_jobs(self): |
| """Schedule new jobs against hosts.""" |
| |
| new_jobs_with_hosts = 0 |
| queue_entries = self.job_query_manager.get_pending_queue_entries( |
| only_hostless=False) |
| unverified_host_jobs = [job for job in queue_entries |
| if not job.is_hostless()] |
| if unverified_host_jobs: |
| for acquisition in self.find_hosts_for_jobs(unverified_host_jobs): |
| self.schedule_host_job(acquisition.host, acquisition.job) |
| self._record_host_assignment(acquisition.host, acquisition.job) |
| new_jobs_with_hosts += 1 |
| metrics.Counter('%s/new_jobs_with_hosts' % _METRICS_PREFIX |
| ).increment_by(new_jobs_with_hosts) |
| |
| num_jobs_without_hosts = (len(unverified_host_jobs) - |
| new_jobs_with_hosts) |
| metrics.Gauge('%s/current_jobs_without_hosts' % _METRICS_PREFIX |
| ).set(num_jobs_without_hosts) |
| |
| metrics.Counter('%s/tick' % _METRICS_PREFIX).increment() |
| |
| @metrics.SecondsTimerDecorator('%s/lease_hosts_duration' % _METRICS_PREFIX) |
| def _lease_hosts_of_frontend_tasks(self): |
| """Lease hosts of tasks scheduled through the frontend.""" |
| # We really don't need to get all the special tasks here, just the ones |
| # without hqes, but reusing the method used by the scheduler ensures |
| # we prioritize the same way. |
| lease_hostnames = [ |
| task.host.hostname for task in |
| self.job_query_manager.get_prioritized_special_tasks( |
| only_tasks_with_leased_hosts=False) |
| if task.queue_entry_id is None and not task.host.leased] |
| # Leasing a leased hosts here shouldn't be a problem: |
| # 1. The only way a host can be leased is if it's been assigned to |
| # an active hqe or another similar frontend task, but doing so will |
| # have already precluded it from the list of tasks returned by the |
| # job_query_manager. |
| # 2. The unleasing is done based on global conditions. Eg: Even if a |
| # task has already leased a host and we lease it again, the |
| # host scheduler won't release the host till both tasks are complete. |
| if lease_hostnames: |
| self.host_query_manager.set_leased( |
| True, hostname__in=lease_hostnames) |
| |
| |
| def acquire_hosts(self, host_jobs): |
| """Override acquire_hosts. |
| |
| This method overrides the method in parent class. |
| It figures out a set of suites that |host_jobs| belong to; |
| and get min_duts requirement for each suite. |
| It pipes min_duts for each suite to rdb. |
| |
| """ |
| parent_job_ids = set([q.job.parent_job_id |
| for q in host_jobs if q.job.parent_job_id]) |
| suite_min_duts = self._suite_recorder.get_min_duts(parent_job_ids) |
| return rdb_lib.acquire_hosts(host_jobs, suite_min_duts) |
| |
| |
| @metrics.SecondsTimerDecorator('%s/tick_time' % _METRICS_PREFIX) |
| def tick(self): |
| logging.info('Calling new tick.') |
| logging.info('Leasing hosts for frontend tasks.') |
| self._lease_hosts_of_frontend_tasks() |
| logging.info('Finding hosts for new jobs.') |
| self._schedule_jobs() |
| logging.info('Releasing unused hosts.') |
| released_hosts = self._release_hosts() |
| logging.info('Updating suite assignment with released hosts') |
| self._suite_recorder.record_release(released_hosts) |
| logging.info('Calling email_manager.') |
| email_manager.manager.send_queued_emails() |
| |
| |
| class DummyHostScheduler(BaseHostScheduler): |
| """A dummy host scheduler that doesn't acquire or release hosts.""" |
| |
| def __init__(self): |
| pass |
| |
| |
| def tick(self): |
| pass |
| |
| |
| def handle_signal(signum, frame): |
| """Sigint handler so we don't crash mid-tick.""" |
| global _shutdown |
| _shutdown = True |
| logging.info("Shutdown request received.") |
| |
| |
| def initialize(testing=False): |
| """Initialize the host scheduler.""" |
| if testing: |
| # Don't import testing utilities unless we're in testing mode, |
| # as the database imports have side effects. |
| from autotest_lib.scheduler import rdb_testing_utils |
| rdb_testing_utils.FileDatabaseHelper().initialize_database_for_testing( |
| db_file_path=rdb_testing_utils.FileDatabaseHelper.DB_FILE) |
| global _db_manager |
| _db_manager = scheduler_lib.ConnectionManager() |
| scheduler_lib.setup_logging( |
| os.environ.get('AUTOTEST_SCHEDULER_LOG_DIR', None), |
| None, timestamped_logfile_prefix='host_scheduler') |
| logging.info("Setting signal handler") |
| signal.signal(signal.SIGINT, handle_signal) |
| signal.signal(signal.SIGTERM, handle_signal) |
| scheduler_models.initialize() |
| |
| |
| def parse_arguments(argv): |
| """ |
| Parse command line arguments |
| |
| @param argv: argument list to parse |
| @returns: parsed arguments. |
| """ |
| parser = argparse.ArgumentParser(description='Host scheduler.') |
| parser.add_argument('--testing', action='store_true', default=False, |
| help='Start the host scheduler in testing mode.') |
| parser.add_argument('--production', |
| help=('Indicate that scheduler is running in production' |
| ' environment and it can use database that is not' |
| ' hosted in localhost. If it is set to False, ' |
| 'scheduler will fail if database is not in ' |
| 'localhost.'), |
| action='store_true', default=False) |
| parser.add_argument( |
| '--lifetime-hours', |
| type=float, |
| default=None, |
| help='If provided, number of hours the scheduler should run for. ' |
| 'At the expiry of this time, the process will exit ' |
| 'gracefully.', |
| ) |
| parser.add_argument( |
| '--metrics-file', |
| help='If provided, drop metrics to this local file instead of ' |
| 'reporting to ts_mon', |
| type=str, |
| default=None, |
| ) |
| options = parser.parse_args(argv) |
| |
| return options |
| |
| |
| def main(): |
| if _monitor_db_host_acquisition: |
| logging.info('Please set inline_host_acquisition=False in the shadow ' |
| 'config before starting the host scheduler.') |
| sys.exit(0) |
| try: |
| options = parse_arguments(sys.argv[1:]) |
| scheduler_lib.check_production_settings(options) |
| |
| # If server database is enabled, check if the server has role |
| # `host_scheduler`. If the server does not have host_scheduler role, |
| # exception will be raised and host scheduler will not continue to run. |
| if server_manager_utils.use_server_db(): |
| server_manager_utils.confirm_server_has_role(hostname='localhost', |
| role='host_scheduler') |
| |
| initialize(options.testing) |
| |
| with ts_mon_config.SetupTsMonGlobalState( |
| 'autotest_host_scheduler', |
| indirect=True, |
| debug_file=options.metrics_file, |
| ): |
| metrics.Counter('%s/start' % _METRICS_PREFIX).increment() |
| process_start_time = time.time() |
| host_scheduler = HostScheduler() |
| minimum_tick_sec = global_config.global_config.get_config_value( |
| 'SCHEDULER', 'host_scheduler_minimum_tick_sec', type=float) |
| while not _shutdown: |
| if _lifetime_expired(options.lifetime_hours, |
| process_start_time): |
| break |
| start = time.time() |
| host_scheduler.tick() |
| curr_tick_sec = time.time() - start |
| if (minimum_tick_sec > curr_tick_sec): |
| time.sleep(minimum_tick_sec - curr_tick_sec) |
| else: |
| time.sleep(0.0001) |
| logging.info('Shutdown request recieved. Bye! Bye!') |
| except server_manager_utils.ServerActionError: |
| # This error is expected when the server is not in primary status |
| # for host-scheduler role. Thus do not send email for it. |
| raise |
| except Exception: |
| metrics.Counter('%s/uncaught_exception' % _METRICS_PREFIX).increment() |
| raise |
| finally: |
| email_manager.manager.send_queued_emails() |
| if _db_manager: |
| _db_manager.disconnect() |
| |
| |
| def _lifetime_expired(lifetime_hours, process_start_time): |
| """Returns True if we've expired the process lifetime, False otherwise. |
| |
| Also sets the global _shutdown so that any background processes also take |
| the cue to exit. |
| """ |
| if lifetime_hours is None: |
| return False |
| if time.time() - process_start_time > lifetime_hours * 3600: |
| logging.info('Process lifetime %0.3f hours exceeded. Shutting down.', |
| lifetime_hours) |
| global _shutdown |
| _shutdown = True |
| return True |
| return False |
| |
| |
| if __name__ == '__main__': |
| main() |
