#!/usr/bin/python
|
#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()
|
