#!/usr/bin/python
|
|
#pylint: disable=C0111
|
|
"""
|
Autotest scheduler
|
"""
|
|
import datetime
|
import functools
|
import logging
|
import optparse
|
import os
|
import signal
|
import sys
|
import time
|
|
import common
|
from autotest_lib.frontend import setup_django_environment
|
|
import django.db
|
|
from autotest_lib.client.common_lib import control_data
|
from autotest_lib.client.common_lib import global_config
|
from autotest_lib.client.common_lib import utils
|
from autotest_lib.frontend.afe import models
|
from autotest_lib.scheduler import agent_task, drone_manager
|
from autotest_lib.scheduler import email_manager, host_scheduler
|
from autotest_lib.scheduler import luciferlib
|
from autotest_lib.scheduler import monitor_db_cleanup, prejob_task
|
from autotest_lib.scheduler import postjob_task
|
from autotest_lib.scheduler import query_managers
|
from autotest_lib.scheduler import scheduler_lib
|
from autotest_lib.scheduler import scheduler_models
|
from autotest_lib.scheduler import scheduler_config
|
from autotest_lib.server import autoserv_utils
|
from autotest_lib.server import system_utils
|
from autotest_lib.server import utils as server_utils
|
from autotest_lib.site_utils import server_manager_utils
|
|
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
|
|
|
PID_FILE_PREFIX = 'monitor_db'
|
|
RESULTS_DIR = '.'
|
AUTOTEST_PATH = os.path.join(os.path.dirname(__file__), '..')
|
|
if os.environ.has_key('AUTOTEST_DIR'):
|
AUTOTEST_PATH = os.environ['AUTOTEST_DIR']
|
AUTOTEST_SERVER_DIR = os.path.join(AUTOTEST_PATH, 'server')
|
AUTOTEST_TKO_DIR = os.path.join(AUTOTEST_PATH, 'tko')
|
|
if AUTOTEST_SERVER_DIR not in sys.path:
|
sys.path.insert(0, AUTOTEST_SERVER_DIR)
|
|
# error message to leave in results dir when an autoserv process disappears
|
# mysteriously
|
_LOST_PROCESS_ERROR = """\
|
Autoserv failed abnormally during execution for this job, probably due to a
|
system error on the Autotest server. Full results may not be available. Sorry.
|
"""
|
|
_db_manager = None
|
_db = None
|
_shutdown = False
|
|
# These 2 globals are replaced for testing
|
_autoserv_directory = autoserv_utils.autoserv_directory
|
_autoserv_path = autoserv_utils.autoserv_path
|
_testing_mode = False
|
_drone_manager = None
|
|
|
def _verify_default_drone_set_exists():
|
if (models.DroneSet.drone_sets_enabled() and
|
not models.DroneSet.default_drone_set_name()):
|
raise scheduler_lib.SchedulerError(
|
'Drone sets are enabled, but no default is set')
|
|
|
def _sanity_check():
|
"""Make sure the configs are consistent before starting the scheduler"""
|
_verify_default_drone_set_exists()
|
|
|
def main():
|
try:
|
try:
|
main_without_exception_handling()
|
except SystemExit:
|
raise
|
except:
|
logging.exception('Exception escaping in monitor_db')
|
raise
|
finally:
|
utils.delete_pid_file_if_exists(PID_FILE_PREFIX)
|
|
|
def main_without_exception_handling():
|
scheduler_lib.setup_logging(
|
os.environ.get('AUTOTEST_SCHEDULER_LOG_DIR', None),
|
os.environ.get('AUTOTEST_SCHEDULER_LOG_NAME', None))
|
usage = 'usage: %prog [options] results_dir'
|
parser = optparse.OptionParser(usage)
|
parser.add_option('--recover-hosts', help='Try to recover dead hosts',
|
action='store_true')
|
parser.add_option('--test', help='Indicate that scheduler is under ' +
|
'test and should use dummy autoserv and no parsing',
|
action='store_true')
|
parser.add_option(
|
'--metrics-file',
|
help='If provided, drop metrics to this local file instead of '
|
'reporting to ts_mon',
|
type=str,
|
default=None,
|
)
|
parser.add_option(
|
'--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_option('--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)
|
(options, args) = parser.parse_args()
|
if len(args) != 1:
|
parser.print_usage()
|
return
|
|
scheduler_lib.check_production_settings(options)
|
|
scheduler_enabled = global_config.global_config.get_config_value(
|
scheduler_config.CONFIG_SECTION, 'enable_scheduler', type=bool)
|
|
if not scheduler_enabled:
|
logging.error("Scheduler not enabled, set enable_scheduler to true in "
|
"the global_config's SCHEDULER section to enable it. "
|
"Exiting.")
|
sys.exit(1)
|
|
global RESULTS_DIR
|
RESULTS_DIR = args[0]
|
|
# Change the cwd while running to avoid issues incase we were launched from
|
# somewhere odd (such as a random NFS home directory of the person running
|
# sudo to launch us as the appropriate user).
|
os.chdir(RESULTS_DIR)
|
|
# This is helpful for debugging why stuff a scheduler launches is
|
# misbehaving.
|
logging.info('os.environ: %s', os.environ)
|
|
if options.test:
|
global _autoserv_path
|
_autoserv_path = 'autoserv_dummy'
|
global _testing_mode
|
_testing_mode = True
|
|
with ts_mon_config.SetupTsMonGlobalState('autotest_scheduler',
|
indirect=True,
|
debug_file=options.metrics_file):
|
try:
|
metrics.Counter('chromeos/autotest/scheduler/start').increment()
|
process_start_time = time.time()
|
initialize()
|
dispatcher = Dispatcher()
|
dispatcher.initialize(recover_hosts=options.recover_hosts)
|
minimum_tick_sec = global_config.global_config.get_config_value(
|
scheduler_config.CONFIG_SECTION, 'minimum_tick_sec', type=float)
|
|
# TODO(crbug.com/837680): Force creating the current user.
|
# This is a dirty hack to work around a race; see bug.
|
models.User.current_user()
|
|
while not _shutdown:
|
if _lifetime_expired(options.lifetime_hours, process_start_time):
|
break
|
|
start = time.time()
|
dispatcher.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)
|
except server_manager_utils.ServerActionError as e:
|
# This error is expected when the server is not in primary status
|
# for scheduler role. Thus do not send email for it.
|
logging.exception(e)
|
except Exception:
|
logging.exception('Uncaught exception, terminating monitor_db.')
|
metrics.Counter('chromeos/autotest/scheduler/uncaught_exception'
|
).increment()
|
|
email_manager.manager.send_queued_emails()
|
_drone_manager.shutdown()
|
_db_manager.disconnect()
|
|
|
def handle_signal(signum, frame):
|
global _shutdown
|
_shutdown = True
|
logging.info("Shutdown request received.")
|
|
|
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
|
|
|
def initialize():
|
logging.info("%s> dispatcher starting", time.strftime("%X %x"))
|
logging.info("My PID is %d", os.getpid())
|
|
if utils.program_is_alive(PID_FILE_PREFIX):
|
logging.critical("monitor_db already running, aborting!")
|
sys.exit(1)
|
utils.write_pid(PID_FILE_PREFIX)
|
|
if _testing_mode:
|
global_config.global_config.override_config_value(
|
scheduler_lib.DB_CONFIG_SECTION, 'database',
|
'stresstest_autotest_web')
|
|
# If server database is enabled, check if the server has role `scheduler`.
|
# If the server does not have scheduler role, exception will be raised and
|
# scheduler will not continue to run.
|
if server_manager_utils.use_server_db():
|
server_manager_utils.confirm_server_has_role(hostname='localhost',
|
role='scheduler')
|
|
os.environ['PATH'] = AUTOTEST_SERVER_DIR + ':' + os.environ['PATH']
|
global _db_manager
|
_db_manager = scheduler_lib.ConnectionManager()
|
global _db
|
_db = _db_manager.get_connection()
|
logging.info("Setting signal handler")
|
signal.signal(signal.SIGINT, handle_signal)
|
signal.signal(signal.SIGTERM, handle_signal)
|
|
initialize_globals()
|
scheduler_models.initialize()
|
|
drone_list = system_utils.get_drones()
|
results_host = global_config.global_config.get_config_value(
|
scheduler_config.CONFIG_SECTION, 'results_host', default='localhost')
|
_drone_manager.initialize(RESULTS_DIR, drone_list, results_host)
|
|
logging.info("Connected! Running...")
|
|
|
def initialize_globals():
|
global _drone_manager
|
_drone_manager = drone_manager.instance()
|
|
|
def _autoserv_command_line(machines, extra_args, job=None, queue_entry=None,
|
verbose=True):
|
"""
|
@returns The autoserv command line as a list of executable + parameters.
|
|
@param machines - string - A machine or comma separated list of machines
|
for the (-m) flag.
|
@param extra_args - list - Additional arguments to pass to autoserv.
|
@param job - Job object - If supplied, -u owner, -l name and client -c or
|
server -s parameters will be added.
|
@param queue_entry - A HostQueueEntry object - If supplied and no Job
|
object was supplied, this will be used to lookup the Job object.
|
"""
|
command = autoserv_utils.autoserv_run_job_command(_autoserv_directory,
|
machines, results_directory=drone_manager.WORKING_DIRECTORY,
|
extra_args=extra_args, job=job, queue_entry=queue_entry,
|
verbose=verbose, in_lab=True)
|
return command
|
|
def _calls_log_tick_msg(func):
|
"""Used to trace functions called by Dispatcher.tick."""
|
@functools.wraps(func)
|
def wrapper(self, *args, **kwargs):
|
self._log_tick_msg('Starting %s' % func.__name__)
|
return func(self, *args, **kwargs)
|
|
return wrapper
|
|
|
class Dispatcher(object):
|
|
|
def __init__(self):
|
self._agents = []
|
self._last_clean_time = time.time()
|
user_cleanup_time = scheduler_config.config.clean_interval_minutes
|
self._periodic_cleanup = monitor_db_cleanup.UserCleanup(
|
_db, user_cleanup_time)
|
self._24hr_upkeep = monitor_db_cleanup.TwentyFourHourUpkeep(
|
_db, _drone_manager)
|
self._host_agents = {}
|
self._queue_entry_agents = {}
|
self._tick_count = 0
|
self._tick_debug = global_config.global_config.get_config_value(
|
scheduler_config.CONFIG_SECTION, 'tick_debug', type=bool,
|
default=False)
|
self._extra_debugging = global_config.global_config.get_config_value(
|
scheduler_config.CONFIG_SECTION, 'extra_debugging', type=bool,
|
default=False)
|
self._inline_host_acquisition = (
|
global_config.global_config.get_config_value(
|
scheduler_config.CONFIG_SECTION,
|
'inline_host_acquisition', type=bool, default=True))
|
|
# If _inline_host_acquisition is set the scheduler will acquire and
|
# release hosts against jobs inline, with the tick. Otherwise the
|
# scheduler will only focus on jobs that already have hosts, and
|
# will not explicitly unlease a host when a job finishes using it.
|
self._job_query_manager = query_managers.AFEJobQueryManager()
|
self._host_scheduler = (host_scheduler.BaseHostScheduler()
|
if self._inline_host_acquisition else
|
host_scheduler.DummyHostScheduler())
|
|
|
def initialize(self, recover_hosts=True):
|
self._periodic_cleanup.initialize()
|
self._24hr_upkeep.initialize()
|
# Execute all actions queued in the cleanup tasks. Scheduler tick will
|
# run a refresh task first. If there is any action in the queue, refresh
|
# will raise an exception.
|
_drone_manager.execute_actions()
|
|
# always recover processes
|
self._recover_processes()
|
|
if recover_hosts:
|
self._recover_hosts()
|
|
|
# TODO(pprabhu) Drop this metric once tick_times has been verified.
|
@metrics.SecondsTimerDecorator(
|
'chromeos/autotest/scheduler/tick_durations/tick')
|
def tick(self):
|
"""
|
This is an altered version of tick() where we keep track of when each
|
major step begins so we can try to figure out where we are using most
|
of the tick time.
|
"""
|
with metrics.RuntimeBreakdownTimer(
|
'chromeos/autotest/scheduler/tick_times') as breakdown_timer:
|
self._log_tick_msg('New tick')
|
system_utils.DroneCache.refresh()
|
|
with breakdown_timer.Step('trigger_refresh'):
|
self._log_tick_msg('Starting _drone_manager.trigger_refresh')
|
_drone_manager.trigger_refresh()
|
with breakdown_timer.Step('schedule_running_host_queue_entries'):
|
self._schedule_running_host_queue_entries()
|
with breakdown_timer.Step('schedule_special_tasks'):
|
self._schedule_special_tasks()
|
with breakdown_timer.Step('schedule_new_jobs'):
|
self._schedule_new_jobs()
|
with breakdown_timer.Step('gather_tick_metrics'):
|
self._gather_tick_metrics()
|
with breakdown_timer.Step('sync_refresh'):
|
self._log_tick_msg('Starting _drone_manager.sync_refresh')
|
_drone_manager.sync_refresh()
|
if luciferlib.is_lucifer_enabled():
|
with breakdown_timer.Step('send_to_lucifer'):
|
self._send_to_lucifer()
|
# _run_cleanup must be called between drone_manager.sync_refresh,
|
# and drone_manager.execute_actions, as sync_refresh will clear the
|
# calls queued in drones. Therefore, any action that calls
|
# drone.queue_call to add calls to the drone._calls, should be after
|
# drone refresh is completed and before
|
# drone_manager.execute_actions at the end of the tick.
|
with breakdown_timer.Step('run_cleanup'):
|
self._run_cleanup()
|
with breakdown_timer.Step('find_aborting'):
|
self._find_aborting()
|
with breakdown_timer.Step('find_aborted_special_tasks'):
|
self._find_aborted_special_tasks()
|
with breakdown_timer.Step('handle_agents'):
|
self._handle_agents()
|
with breakdown_timer.Step('host_scheduler_tick'):
|
self._log_tick_msg('Starting _host_scheduler.tick')
|
self._host_scheduler.tick()
|
with breakdown_timer.Step('drones_execute_actions'):
|
self._log_tick_msg('Starting _drone_manager.execute_actions')
|
_drone_manager.execute_actions()
|
with breakdown_timer.Step('send_queued_emails'):
|
self._log_tick_msg(
|
'Starting email_manager.manager.send_queued_emails')
|
email_manager.manager.send_queued_emails()
|
with breakdown_timer.Step('db_reset_queries'):
|
self._log_tick_msg('Starting django.db.reset_queries')
|
django.db.reset_queries()
|
|
self._tick_count += 1
|
metrics.Counter('chromeos/autotest/scheduler/tick').increment()
|
|
|
@_calls_log_tick_msg
|
def _run_cleanup(self):
|
self._periodic_cleanup.run_cleanup_maybe()
|
self._24hr_upkeep.run_cleanup_maybe()
|
|
|
def _gather_tick_metrics(self):
|
"""Gather metrics during tick, after all tasks have been scheduled."""
|
metrics.Gauge(
|
'chromeos/autotest/scheduler/agent_count'
|
).set(len(self._agents))
|
|
|
def _register_agent_for_ids(self, agent_dict, object_ids, agent):
|
for object_id in object_ids:
|
agent_dict.setdefault(object_id, set()).add(agent)
|
|
|
def _unregister_agent_for_ids(self, agent_dict, object_ids, agent):
|
for object_id in object_ids:
|
assert object_id in agent_dict
|
agent_dict[object_id].remove(agent)
|
# If an ID has no more active agent associated, there is no need to
|
# keep it in the dictionary. Otherwise, scheduler will keep an
|
# unnecessarily big dictionary until being restarted.
|
if not agent_dict[object_id]:
|
agent_dict.pop(object_id)
|
|
|
def add_agent_task(self, agent_task):
|
"""
|
Creates and adds an agent to the dispatchers list.
|
|
In creating the agent we also pass on all the queue_entry_ids and
|
host_ids from the special agent task. For every agent we create, we
|
add it to 1. a dict against the queue_entry_ids given to it 2. A dict
|
against the host_ids given to it. So theoritically, a host can have any
|
number of agents associated with it, and each of them can have any
|
special agent task, though in practice we never see > 1 agent/task per
|
host at any time.
|
|
@param agent_task: A SpecialTask for the agent to manage.
|
"""
|
if luciferlib.is_enabled_for('STARTING'):
|
# TODO(crbug.com/810141): Transition code. After running at
|
# STARTING for a while, these tasks should no longer exist.
|
if (isinstance(agent_task, postjob_task.GatherLogsTask)
|
# TODO(crbug.com/811877): Don't skip split HQE parsing.
|
or (isinstance(agent_task, postjob_task.FinalReparseTask)
|
and not luciferlib.is_split_job(
|
agent_task.queue_entries[0].id))):
|
return
|
if isinstance(agent_task, AbstractQueueTask):
|
# If Lucifer already owns the job, ignore the agent.
|
if luciferlib.is_lucifer_owned_by_id(agent_task.job.id):
|
return
|
# If the job isn't started yet, let Lucifer own it.
|
if not agent_task.started:
|
return
|
# Otherwise, this is a STARTING job that Autotest owned
|
# before Lucifer was enabled for STARTING. Allow the
|
# scheduler to recover the agent task normally.
|
|
agent = Agent(agent_task)
|
self._agents.append(agent)
|
agent.dispatcher = self
|
self._register_agent_for_ids(self._host_agents, agent.host_ids, agent)
|
self._register_agent_for_ids(self._queue_entry_agents,
|
agent.queue_entry_ids, agent)
|
|
|
def get_agents_for_entry(self, queue_entry):
|
"""
|
Find agents corresponding to the specified queue_entry.
|
"""
|
return list(self._queue_entry_agents.get(queue_entry.id, set()))
|
|
|
def host_has_agent(self, host):
|
"""
|
Determine if there is currently an Agent present using this host.
|
"""
|
return bool(self._host_agents.get(host.id, None))
|
|
|
def remove_agent(self, agent):
|
self._agents.remove(agent)
|
self._unregister_agent_for_ids(self._host_agents, agent.host_ids,
|
agent)
|
self._unregister_agent_for_ids(self._queue_entry_agents,
|
agent.queue_entry_ids, agent)
|
|
|
def _host_has_scheduled_special_task(self, host):
|
return bool(models.SpecialTask.objects.filter(host__id=host.id,
|
is_active=False,
|
is_complete=False))
|
|
|
def _recover_processes(self):
|
agent_tasks = self._create_recovery_agent_tasks()
|
self._register_pidfiles(agent_tasks)
|
_drone_manager.refresh()
|
self._recover_tasks(agent_tasks)
|
self._recover_pending_entries()
|
self._check_for_unrecovered_verifying_entries()
|
self._reverify_remaining_hosts()
|
# reinitialize drones after killing orphaned processes, since they can
|
# leave around files when they die
|
_drone_manager.execute_actions()
|
_drone_manager.reinitialize_drones()
|
|
|
def _create_recovery_agent_tasks(self):
|
return (self._get_queue_entry_agent_tasks()
|
+ self._get_special_task_agent_tasks(is_active=True))
|
|
|
def _get_queue_entry_agent_tasks(self):
|
"""
|
Get agent tasks for all hqe in the specified states.
|
|
Loosely this translates to taking a hqe in one of the specified states,
|
say parsing, and getting an AgentTask for it, like the FinalReparseTask,
|
through _get_agent_task_for_queue_entry. Each queue entry can only have
|
one agent task at a time, but there might be multiple queue entries in
|
the group.
|
|
@return: A list of AgentTasks.
|
"""
|
# host queue entry statuses handled directly by AgentTasks
|
# (Verifying is handled through SpecialTasks, so is not
|
# listed here)
|
statuses = (models.HostQueueEntry.Status.STARTING,
|
models.HostQueueEntry.Status.RUNNING,
|
models.HostQueueEntry.Status.GATHERING,
|
models.HostQueueEntry.Status.PARSING)
|
status_list = ','.join("'%s'" % status for status in statuses)
|
queue_entries = scheduler_models.HostQueueEntry.fetch(
|
where='status IN (%s)' % status_list)
|
|
agent_tasks = []
|
used_queue_entries = set()
|
hqe_count_by_status = {}
|
for entry in queue_entries:
|
try:
|
hqe_count_by_status[entry.status] = (
|
hqe_count_by_status.get(entry.status, 0) + 1)
|
if self.get_agents_for_entry(entry):
|
# already being handled
|
continue
|
if entry in used_queue_entries:
|
# already picked up by a synchronous job
|
continue
|
try:
|
agent_task = self._get_agent_task_for_queue_entry(entry)
|
except scheduler_lib.SchedulerError:
|
# Probably being handled by lucifer crbug.com/809773
|
continue
|
agent_tasks.append(agent_task)
|
used_queue_entries.update(agent_task.queue_entries)
|
except scheduler_lib.MalformedRecordError as e:
|
logging.exception('Skipping agent task for a malformed hqe.')
|
# TODO(akeshet): figure out a way to safely permanently discard
|
# this errant HQE. It appears that calling entry.abort() is not
|
# sufficient, as that already makes some assumptions about
|
# record sanity that may be violated. See crbug.com/739530 for
|
# context.
|
m = 'chromeos/autotest/scheduler/skipped_malformed_hqe'
|
metrics.Counter(m).increment()
|
|
for status, count in hqe_count_by_status.iteritems():
|
metrics.Gauge(
|
'chromeos/autotest/scheduler/active_host_queue_entries'
|
).set(count, fields={'status': status})
|
|
return agent_tasks
|
|
|
def _get_special_task_agent_tasks(self, is_active=False):
|
special_tasks = models.SpecialTask.objects.filter(
|
is_active=is_active, is_complete=False)
|
agent_tasks = []
|
for task in special_tasks:
|
try:
|
agent_tasks.append(self._get_agent_task_for_special_task(task))
|
except scheduler_lib.MalformedRecordError as e:
|
logging.exception('Skipping agent task for malformed special '
|
'task.')
|
m = 'chromeos/autotest/scheduler/skipped_malformed_special_task'
|
metrics.Counter(m).increment()
|
return agent_tasks
|
|
|
def _get_agent_task_for_queue_entry(self, queue_entry):
|
"""
|
Construct an AgentTask instance for the given active HostQueueEntry.
|
|
@param queue_entry: a HostQueueEntry
|
@return: an AgentTask to run the queue entry
|
"""
|
task_entries = queue_entry.job.get_group_entries(queue_entry)
|
self._check_for_duplicate_host_entries(task_entries)
|
|
if queue_entry.status in (models.HostQueueEntry.Status.STARTING,
|
models.HostQueueEntry.Status.RUNNING):
|
if queue_entry.is_hostless():
|
return HostlessQueueTask(queue_entry=queue_entry)
|
return QueueTask(queue_entries=task_entries)
|
if queue_entry.status == models.HostQueueEntry.Status.GATHERING:
|
return postjob_task.GatherLogsTask(queue_entries=task_entries)
|
if queue_entry.status == models.HostQueueEntry.Status.PARSING:
|
return postjob_task.FinalReparseTask(queue_entries=task_entries)
|
|
raise scheduler_lib.MalformedRecordError(
|
'_get_agent_task_for_queue_entry got entry with '
|
'invalid status %s: %s' % (queue_entry.status, queue_entry))
|
|
|
def _check_for_duplicate_host_entries(self, task_entries):
|
non_host_statuses = {models.HostQueueEntry.Status.PARSING}
|
for task_entry in task_entries:
|
using_host = (task_entry.host is not None
|
and task_entry.status not in non_host_statuses)
|
if using_host:
|
self._assert_host_has_no_agent(task_entry)
|
|
|
def _assert_host_has_no_agent(self, entry):
|
"""
|
@param entry: a HostQueueEntry or a SpecialTask
|
"""
|
if self.host_has_agent(entry.host):
|
agent = tuple(self._host_agents.get(entry.host.id))[0]
|
raise scheduler_lib.MalformedRecordError(
|
'While scheduling %s, host %s already has a host agent %s'
|
% (entry, entry.host, agent.task))
|
|
|
def _get_agent_task_for_special_task(self, special_task):
|
"""
|
Construct an AgentTask class to run the given SpecialTask and add it
|
to this dispatcher.
|
|
A special task is created through schedule_special_tasks, but only if
|
the host doesn't already have an agent. This happens through
|
add_agent_task. All special agent tasks are given a host on creation,
|
and a Null hqe. To create a SpecialAgentTask object, you need a
|
models.SpecialTask. If the SpecialTask used to create a SpecialAgentTask
|
object contains a hqe it's passed on to the special agent task, which
|
creates a HostQueueEntry and saves it as it's queue_entry.
|
|
@param special_task: a models.SpecialTask instance
|
@returns an AgentTask to run this SpecialTask
|
"""
|
self._assert_host_has_no_agent(special_task)
|
|
special_agent_task_classes = (prejob_task.CleanupTask,
|
prejob_task.VerifyTask,
|
prejob_task.RepairTask,
|
prejob_task.ResetTask,
|
prejob_task.ProvisionTask)
|
|
for agent_task_class in special_agent_task_classes:
|
if agent_task_class.TASK_TYPE == special_task.task:
|
return agent_task_class(task=special_task)
|
|
raise scheduler_lib.MalformedRecordError(
|
'No AgentTask class for task', str(special_task))
|
|
|
def _register_pidfiles(self, agent_tasks):
|
for agent_task in agent_tasks:
|
agent_task.register_necessary_pidfiles()
|
|
|
def _recover_tasks(self, agent_tasks):
|
orphans = _drone_manager.get_orphaned_autoserv_processes()
|
|
for agent_task in agent_tasks:
|
agent_task.recover()
|
if agent_task.monitor and agent_task.monitor.has_process():
|
orphans.discard(agent_task.monitor.get_process())
|
self.add_agent_task(agent_task)
|
|
self._check_for_remaining_orphan_processes(orphans)
|
|
|
def _get_unassigned_entries(self, status):
|
for entry in scheduler_models.HostQueueEntry.fetch(where="status = '%s'"
|
% status):
|
if entry.status == status and not self.get_agents_for_entry(entry):
|
# The status can change during iteration, e.g., if job.run()
|
# sets a group of queue entries to Starting
|
yield entry
|
|
|
def _check_for_remaining_orphan_processes(self, orphans):
|
m = 'chromeos/autotest/errors/unrecovered_orphan_processes'
|
metrics.Gauge(m).set(len(orphans))
|
|
if not orphans:
|
return
|
subject = 'Unrecovered orphan autoserv processes remain'
|
message = '\n'.join(str(process) for process in orphans)
|
die_on_orphans = global_config.global_config.get_config_value(
|
scheduler_config.CONFIG_SECTION, 'die_on_orphans', type=bool)
|
|
if die_on_orphans:
|
raise RuntimeError(subject + '\n' + message)
|
|
|
def _recover_pending_entries(self):
|
for entry in self._get_unassigned_entries(
|
models.HostQueueEntry.Status.PENDING):
|
logging.info('Recovering Pending entry %s', entry)
|
try:
|
entry.on_pending()
|
except scheduler_lib.MalformedRecordError as e:
|
logging.exception(
|
'Skipping agent task for malformed special task.')
|
m = 'chromeos/autotest/scheduler/skipped_malformed_special_task'
|
metrics.Counter(m).increment()
|
|
|
def _check_for_unrecovered_verifying_entries(self):
|
# Verify is replaced by Reset.
|
queue_entries = scheduler_models.HostQueueEntry.fetch(
|
where='status = "%s"' % models.HostQueueEntry.Status.RESETTING)
|
for queue_entry in queue_entries:
|
special_tasks = models.SpecialTask.objects.filter(
|
task__in=(models.SpecialTask.Task.CLEANUP,
|
models.SpecialTask.Task.VERIFY,
|
models.SpecialTask.Task.RESET),
|
queue_entry__id=queue_entry.id,
|
is_complete=False)
|
if special_tasks.count() == 0:
|
logging.error('Unrecovered Resetting host queue entry: %s. ',
|
str(queue_entry))
|
# Essentially this host queue entry was set to be Verifying
|
# however no special task exists for entry. This occurs if the
|
# scheduler dies between changing the status and creating the
|
# special task. By setting it to queued, the job can restart
|
# from the beginning and proceed correctly. This is much more
|
# preferable than having monitor_db not launching.
|
logging.info('Setting host status for %s to Ready',
|
str(queue_entry.host))
|
# Let's at least run a cleanup/reset before reusing this DUT.
|
queue_entry.host.update_field('dirty', 1)
|
queue_entry.host.set_status(models.Host.Status.READY)
|
logging.info('Setting status for HQE %s to Queued.',
|
str(queue_entry))
|
queue_entry.set_status('Queued')
|
|
|
@_calls_log_tick_msg
|
def _schedule_special_tasks(self):
|
"""
|
Execute queued SpecialTasks that are ready to run on idle hosts.
|
|
Special tasks include PreJobTasks like verify, reset and cleanup.
|
They are created through _schedule_new_jobs and associated with a hqe
|
This method translates SpecialTasks to the appropriate AgentTask and
|
adds them to the dispatchers agents list, so _handle_agents can execute
|
them.
|
"""
|
# When the host scheduler is responsible for acquisition we only want
|
# to run tasks with leased hosts. All hqe tasks will already have
|
# leased hosts, and we don't want to run frontend tasks till the host
|
# scheduler has vetted the assignment. Note that this doesn't include
|
# frontend tasks with hosts leased by other active hqes.
|
for task in self._job_query_manager.get_prioritized_special_tasks(
|
only_tasks_with_leased_hosts=not self._inline_host_acquisition):
|
if self.host_has_agent(task.host):
|
continue
|
try:
|
self.add_agent_task(self._get_agent_task_for_special_task(task))
|
except scheduler_lib.MalformedRecordError:
|
logging.exception('Skipping schedule for malformed '
|
'special task.')
|
m = 'chromeos/autotest/scheduler/skipped_schedule_special_task'
|
metrics.Counter(m).increment()
|
|
|
def _reverify_remaining_hosts(self):
|
# recover active hosts that have not yet been recovered, although this
|
# should never happen
|
message = ('Recovering active host %s - this probably indicates a '
|
'scheduler bug')
|
self._reverify_hosts_where(
|
"status IN ('Repairing', 'Verifying', 'Cleaning', 'Provisioning')",
|
print_message=message)
|
|
|
DEFAULT_REQUESTED_BY_USER_ID = 1
|
|
|
def _reverify_hosts_where(self, where,
|
print_message='Reverifying host %s'):
|
full_where = 'locked = 0 AND invalid = 0 AND %s' % where
|
for host in scheduler_models.Host.fetch(where=full_where):
|
if self.host_has_agent(host):
|
# host has already been recovered in some way
|
continue
|
if self._host_has_scheduled_special_task(host):
|
# host will have a special task scheduled on the next cycle
|
continue
|
if host.shard_id is not None and not server_utils.is_shard():
|
# I am master and host is owned by a shard, ignore it.
|
continue
|
if print_message:
|
logging.error(print_message, host.hostname)
|
try:
|
user = models.User.objects.get(login='autotest_system')
|
except models.User.DoesNotExist:
|
user = models.User.objects.get(
|
id=self.DEFAULT_REQUESTED_BY_USER_ID)
|
models.SpecialTask.objects.create(
|
task=models.SpecialTask.Task.RESET,
|
host=models.Host.objects.get(id=host.id),
|
requested_by=user)
|
|
|
def _recover_hosts(self):
|
# recover "Repair Failed" hosts
|
message = 'Reverifying dead host %s'
|
self._reverify_hosts_where("status = 'Repair Failed'",
|
print_message=message)
|
|
|
def _refresh_pending_queue_entries(self):
|
"""
|
Lookup the pending HostQueueEntries and call our HostScheduler
|
refresh() method given that list. Return the list.
|
|
@returns A list of pending HostQueueEntries sorted in priority order.
|
"""
|
queue_entries = self._job_query_manager.get_pending_queue_entries(
|
only_hostless=not self._inline_host_acquisition)
|
if not queue_entries:
|
return []
|
return queue_entries
|
|
|
def _schedule_hostless_job(self, queue_entry):
|
"""Schedule a hostless (suite) job.
|
|
@param queue_entry: The queue_entry representing the hostless job.
|
"""
|
if not luciferlib.is_enabled_for('STARTING'):
|
self.add_agent_task(HostlessQueueTask(queue_entry))
|
|
# Need to set execution_subdir before setting the status:
|
# After a restart of the scheduler, agents will be restored for HQEs in
|
# Starting, Running, Gathering, Parsing or Archiving. To do this, the
|
# execution_subdir is needed. Therefore it must be set before entering
|
# one of these states.
|
# Otherwise, if the scheduler was interrupted between setting the status
|
# and the execution_subdir, upon it's restart restoring agents would
|
# fail.
|
# Is there a way to get a status in one of these states without going
|
# through this code? Following cases are possible:
|
# - If it's aborted before being started:
|
# active bit will be 0, so there's nothing to parse, it will just be
|
# set to completed by _find_aborting. Critical statuses are skipped.
|
# - If it's aborted or it fails after being started:
|
# It was started, so this code was executed.
|
queue_entry.update_field('execution_subdir', 'hostless')
|
queue_entry.set_status(models.HostQueueEntry.Status.STARTING)
|
|
|
def _schedule_host_job(self, host, queue_entry):
|
"""Schedules a job on the given host.
|
|
1. Assign the host to the hqe, if it isn't already assigned.
|
2. Create a SpecialAgentTask for the hqe.
|
3. Activate the hqe.
|
|
@param queue_entry: The job to schedule.
|
@param host: The host to schedule the job on.
|
"""
|
if self.host_has_agent(host):
|
host_agent_task = list(self._host_agents.get(host.id))[0].task
|
else:
|
self._host_scheduler.schedule_host_job(host, queue_entry)
|
|
|
@_calls_log_tick_msg
|
def _schedule_new_jobs(self):
|
"""
|
Find any new HQEs and call schedule_pre_job_tasks for it.
|
|
This involves setting the status of the HQE and creating a row in the
|
db corresponding the the special task, through
|
scheduler_models._queue_special_task. The new db row is then added as
|
an agent to the dispatcher through _schedule_special_tasks and
|
scheduled for execution on the drone through _handle_agents.
|
"""
|
queue_entries = self._refresh_pending_queue_entries()
|
|
key = 'scheduler.jobs_per_tick'
|
new_hostless_jobs = 0
|
new_jobs_with_hosts = 0
|
new_jobs_need_hosts = 0
|
host_jobs = []
|
logging.debug('Processing %d queue_entries', len(queue_entries))
|
|
for queue_entry in queue_entries:
|
if queue_entry.is_hostless():
|
self._schedule_hostless_job(queue_entry)
|
new_hostless_jobs = new_hostless_jobs + 1
|
else:
|
host_jobs.append(queue_entry)
|
new_jobs_need_hosts = new_jobs_need_hosts + 1
|
|
metrics.Counter(
|
'chromeos/autotest/scheduler/scheduled_jobs_hostless'
|
).increment_by(new_hostless_jobs)
|
|
if not host_jobs:
|
return
|
|
if not self._inline_host_acquisition:
|
# In this case, host_scheduler is responsible for scheduling
|
# host_jobs. Scheduling the jobs ourselves can lead to DB corruption
|
# since host_scheduler assumes it is the single process scheduling
|
# host jobs.
|
metrics.Gauge(
|
'chromeos/autotest/errors/scheduler/unexpected_host_jobs').set(
|
len(host_jobs))
|
return
|
|
jobs_with_hosts = self._host_scheduler.find_hosts_for_jobs(host_jobs)
|
for host_assignment in jobs_with_hosts:
|
self._schedule_host_job(host_assignment.host, host_assignment.job)
|
new_jobs_with_hosts = new_jobs_with_hosts + 1
|
|
metrics.Counter(
|
'chromeos/autotest/scheduler/scheduled_jobs_with_hosts'
|
).increment_by(new_jobs_with_hosts)
|
|
|
@_calls_log_tick_msg
|
def _send_to_lucifer(self):
|
"""
|
Hand off ownership of a job to lucifer component.
|
"""
|
self._send_starting_to_lucifer()
|
self._send_parsing_to_lucifer()
|
|
|
# TODO(crbug.com/748234): This is temporary to enable toggling
|
# lucifer rollouts with an option.
|
def _send_starting_to_lucifer(self):
|
Status = models.HostQueueEntry.Status
|
queue_entries_qs = (models.HostQueueEntry.objects
|
.filter(status=Status.STARTING))
|
for queue_entry in queue_entries_qs:
|
if self.get_agents_for_entry(queue_entry):
|
continue
|
job = queue_entry.job
|
if luciferlib.is_lucifer_owned(job):
|
continue
|
try:
|
drone = luciferlib.spawn_starting_job_handler(
|
manager=_drone_manager,
|
job=job)
|
except Exception:
|
logging.exception('Error when sending job to Lucifer')
|
models.HostQueueEntry.abort_host_queue_entries(
|
job.hostqueueentry_set.all())
|
else:
|
models.JobHandoff.objects.create(
|
job=job, drone=drone.hostname())
|
|
|
# TODO(crbug.com/748234): This is temporary to enable toggling
|
# lucifer rollouts with an option.
|
def _send_parsing_to_lucifer(self):
|
Status = models.HostQueueEntry.Status
|
queue_entries_qs = (models.HostQueueEntry.objects
|
.filter(status=Status.PARSING))
|
for queue_entry in queue_entries_qs:
|
# If this HQE already has an agent, let monitor_db continue
|
# owning it.
|
if self.get_agents_for_entry(queue_entry):
|
continue
|
job = queue_entry.job
|
if luciferlib.is_lucifer_owned(job):
|
continue
|
# TODO(crbug.com/811877): Ignore split HQEs.
|
if luciferlib.is_split_job(queue_entry.id):
|
continue
|
task = postjob_task.PostJobTask(
|
[queue_entry], log_file_name='/dev/null')
|
pidfile_id = task._autoserv_monitor.pidfile_id
|
autoserv_exit = task._autoserv_monitor.exit_code()
|
try:
|
drone = luciferlib.spawn_parsing_job_handler(
|
manager=_drone_manager,
|
job=job,
|
autoserv_exit=autoserv_exit,
|
pidfile_id=pidfile_id)
|
models.JobHandoff.objects.create(job=job,
|
drone=drone.hostname())
|
except drone_manager.DroneManagerError as e:
|
logging.warning(
|
'Fail to get drone for job %s, skipping lucifer. Error: %s',
|
job.id, e)
|
|
|
|
@_calls_log_tick_msg
|
def _schedule_running_host_queue_entries(self):
|
"""
|
Adds agents to the dispatcher.
|
|
Any AgentTask, like the QueueTask, is wrapped in an Agent. The
|
QueueTask for example, will have a job with a control file, and
|
the agent will have methods that poll, abort and check if the queue
|
task is finished. The dispatcher runs the agent_task, as well as
|
other agents in it's _agents member, through _handle_agents, by
|
calling the Agents tick().
|
|
This method creates an agent for each HQE in one of (starting, running,
|
gathering, parsing) states, and adds it to the dispatcher so
|
it is handled by _handle_agents.
|
"""
|
for agent_task in self._get_queue_entry_agent_tasks():
|
self.add_agent_task(agent_task)
|
|
|
@_calls_log_tick_msg
|
def _find_aborting(self):
|
"""
|
Looks through the afe_host_queue_entries for an aborted entry.
|
|
The aborted bit is set on an HQE in many ways, the most common
|
being when a user requests an abort through the frontend, which
|
results in an rpc from the afe to abort_host_queue_entries.
|
"""
|
jobs_to_stop = set()
|
for entry in scheduler_models.HostQueueEntry.fetch(
|
where='aborted=1 and complete=0'):
|
if (luciferlib.is_enabled_for('STARTING')
|
and luciferlib.is_lucifer_owned_by_id(entry.job.id)):
|
continue
|
|
# If the job is running on a shard, let the shard handle aborting
|
# it and sync back the right status.
|
if entry.job.shard_id is not None and not server_utils.is_shard():
|
# Due to crbug.com/894162, we abort jobs that 1hr beyond
|
# timeout on master.
|
create_on = time.mktime(entry.job.created_on.timetuple())
|
wait_threshold = entry.job.timeout_mins * 60 + 3600
|
abort_anyway = wait_threshold < time.time() - create_on
|
if abort_anyway:
|
logging.info('Aborting %s on master due to '
|
'the job 1 hour beyond timeout.', entry)
|
else:
|
logging.info('Waiting for shard %s to abort hqe %s',
|
entry.job.shard_id, entry)
|
continue
|
|
logging.info('Aborting %s', entry)
|
|
# The task would have started off with both is_complete and
|
# is_active = False. Aborted tasks are neither active nor complete.
|
# For all currently active tasks this will happen through the agent,
|
# but we need to manually update the special tasks that haven't
|
# started yet, because they don't have agents.
|
models.SpecialTask.objects.filter(is_active=False,
|
queue_entry_id=entry.id).update(is_complete=True)
|
|
for agent in self.get_agents_for_entry(entry):
|
agent.abort()
|
entry.abort(self)
|
jobs_to_stop.add(entry.job)
|
logging.debug('Aborting %d jobs this tick.', len(jobs_to_stop))
|
for job in jobs_to_stop:
|
job.stop_if_necessary()
|
|
|
@_calls_log_tick_msg
|
def _find_aborted_special_tasks(self):
|
"""
|
Find SpecialTasks that have been marked for abortion.
|
|
Poll the database looking for SpecialTasks that are active
|
and have been marked for abortion, then abort them.
|
"""
|
|
# The completed and active bits are very important when it comes
|
# to scheduler correctness. The active bit is set through the prolog
|
# of a special task, and reset through the cleanup method of the
|
# SpecialAgentTask. The cleanup is called both through the abort and
|
# epilog. The complete bit is set in several places, and in general
|
# a hanging job will have is_active=1 is_complete=0, while a special
|
# task which completed will have is_active=0 is_complete=1. To check
|
# aborts we directly check active because the complete bit is set in
|
# several places, including the epilog of agent tasks.
|
aborted_tasks = models.SpecialTask.objects.filter(is_active=True,
|
is_aborted=True)
|
for task in aborted_tasks:
|
# There are 2 ways to get the agent associated with a task,
|
# through the host and through the hqe. A special task
|
# always needs a host, but doesn't always need a hqe.
|
for agent in self._host_agents.get(task.host.id, []):
|
if isinstance(agent.task, agent_task.SpecialAgentTask):
|
|
# The epilog preforms critical actions such as
|
# queueing the next SpecialTask, requeuing the
|
# hqe etc, however it doesn't actually kill the
|
# monitor process and set the 'done' bit. Epilogs
|
# assume that the job failed, and that the monitor
|
# process has already written an exit code. The
|
# done bit is a necessary condition for
|
# _handle_agents to schedule any more special
|
# tasks against the host, and it must be set
|
# in addition to is_active, is_complete and success.
|
agent.task.epilog()
|
agent.task.abort()
|
|
|
def _can_start_agent(self, agent, have_reached_limit):
|
# always allow zero-process agents to run
|
if agent.task.num_processes == 0:
|
return True
|
# don't allow any nonzero-process agents to run after we've reached a
|
# limit (this avoids starvation of many-process agents)
|
if have_reached_limit:
|
return False
|
# total process throttling
|
max_runnable_processes = _drone_manager.max_runnable_processes(
|
agent.task.owner_username,
|
agent.task.get_drone_hostnames_allowed())
|
if agent.task.num_processes > max_runnable_processes:
|
return False
|
return True
|
|
|
@_calls_log_tick_msg
|
def _handle_agents(self):
|
"""
|
Handles agents of the dispatcher.
|
|
Appropriate Agents are added to the dispatcher through
|
_schedule_running_host_queue_entries. These agents each
|
have a task. This method runs the agents task through
|
agent.tick() leading to:
|
agent.start
|
prolog -> AgentTasks prolog
|
For each queue entry:
|
sets host status/status to Running
|
set started_on in afe_host_queue_entries
|
run -> AgentTasks run
|
Creates PidfileRunMonitor
|
Queues the autoserv command line for this AgentTask
|
via the drone manager. These commands are executed
|
through the drone managers execute actions.
|
poll -> AgentTasks/BaseAgentTask poll
|
checks the monitors exit_code.
|
Executes epilog if task is finished.
|
Executes AgentTasks _finish_task
|
finish_task is usually responsible for setting the status
|
of the HQE/host, and updating it's active and complete fileds.
|
|
agent.is_done
|
Removed the agent from the dispatchers _agents queue.
|
Is_done checks the finished bit on the agent, that is
|
set based on the Agents task. During the agents poll
|
we check to see if the monitor process has exited in
|
it's finish method, and set the success member of the
|
task based on this exit code.
|
"""
|
num_started_this_tick = 0
|
num_finished_this_tick = 0
|
have_reached_limit = False
|
# iterate over copy, so we can remove agents during iteration
|
logging.debug('Handling %d Agents', len(self._agents))
|
for agent in list(self._agents):
|
self._log_extra_msg('Processing Agent with Host Ids: %s and '
|
'queue_entry ids:%s' % (agent.host_ids,
|
agent.queue_entry_ids))
|
if not agent.started:
|
if not self._can_start_agent(agent, have_reached_limit):
|
have_reached_limit = True
|
logging.debug('Reached Limit of allowed running Agents.')
|
continue
|
num_started_this_tick += agent.task.num_processes
|
self._log_extra_msg('Starting Agent')
|
agent.tick()
|
self._log_extra_msg('Agent tick completed.')
|
if agent.is_done():
|
num_finished_this_tick += agent.task.num_processes
|
self._log_extra_msg("Agent finished")
|
self.remove_agent(agent)
|
|
metrics.Counter(
|
'chromeos/autotest/scheduler/agent_processes_started'
|
).increment_by(num_started_this_tick)
|
metrics.Counter(
|
'chromeos/autotest/scheduler/agent_processes_finished'
|
).increment_by(num_finished_this_tick)
|
num_agent_processes = _drone_manager.total_running_processes()
|
metrics.Gauge(
|
'chromeos/autotest/scheduler/agent_processes'
|
).set(num_agent_processes)
|
logging.info('%d running processes. %d added this tick.',
|
num_agent_processes, num_started_this_tick)
|
|
|
def _log_tick_msg(self, msg):
|
if self._tick_debug:
|
logging.debug(msg)
|
|
|
def _log_extra_msg(self, msg):
|
if self._extra_debugging:
|
logging.debug(msg)
|
|
|
class Agent(object):
|
"""
|
An agent for use by the Dispatcher class to perform a task. An agent wraps
|
around an AgentTask mainly to associate the AgentTask with the queue_entry
|
and host ids.
|
|
The following methods are required on all task objects:
|
poll() - Called periodically to let the task check its status and
|
update its internal state. If the task succeeded.
|
is_done() - Returns True if the task is finished.
|
abort() - Called when an abort has been requested. The task must
|
set its aborted attribute to True if it actually aborted.
|
|
The following attributes are required on all task objects:
|
aborted - bool, True if this task was aborted.
|
success - bool, True if this task succeeded.
|
queue_entry_ids - A sequence of HostQueueEntry ids this task handles.
|
host_ids - A sequence of Host ids this task represents.
|
"""
|
|
|
def __init__(self, task):
|
"""
|
@param task: An instance of an AgentTask.
|
"""
|
self.task = task
|
|
# This is filled in by Dispatcher.add_agent()
|
self.dispatcher = None
|
|
self.queue_entry_ids = task.queue_entry_ids
|
self.host_ids = task.host_ids
|
|
self.started = False
|
self.finished = False
|
|
|
def tick(self):
|
self.started = True
|
if not self.finished:
|
self.task.poll()
|
if self.task.is_done():
|
self.finished = True
|
|
|
def is_done(self):
|
return self.finished
|
|
|
def abort(self):
|
if self.task:
|
self.task.abort()
|
if self.task.aborted:
|
# tasks can choose to ignore aborts
|
self.finished = True
|
|
|
class AbstractQueueTask(agent_task.AgentTask, agent_task.TaskWithJobKeyvals):
|
"""
|
Common functionality for QueueTask and HostlessQueueTask
|
"""
|
def __init__(self, queue_entries):
|
super(AbstractQueueTask, self).__init__()
|
self.job = queue_entries[0].job
|
self.queue_entries = queue_entries
|
|
|
def _keyval_path(self):
|
return os.path.join(self._working_directory(), self._KEYVAL_FILE)
|
|
|
def _write_control_file(self, execution_path):
|
control_path = _drone_manager.attach_file_to_execution(
|
execution_path, self.job.control_file)
|
return control_path
|
|
|
# TODO: Refactor into autoserv_utils. crbug.com/243090
|
def _command_line(self):
|
execution_path = self.queue_entries[0].execution_path()
|
control_path = self._write_control_file(execution_path)
|
hostnames = ','.join(entry.host.hostname
|
for entry in self.queue_entries
|
if not entry.is_hostless())
|
|
execution_tag = self.queue_entries[0].execution_tag()
|
params = _autoserv_command_line(
|
hostnames,
|
['-P', execution_tag, '-n',
|
_drone_manager.absolute_path(control_path)],
|
job=self.job, verbose=False)
|
|
return params
|
|
|
@property
|
def num_processes(self):
|
return len(self.queue_entries)
|
|
|
@property
|
def owner_username(self):
|
return self.job.owner
|
|
|
def _working_directory(self):
|
return self._get_consistent_execution_path(self.queue_entries)
|
|
|
def prolog(self):
|
queued_key, queued_time = self._job_queued_keyval(self.job)
|
keyval_dict = self.job.keyval_dict()
|
keyval_dict[queued_key] = queued_time
|
self._write_keyvals_before_job(keyval_dict)
|
for queue_entry in self.queue_entries:
|
queue_entry.set_status(models.HostQueueEntry.Status.RUNNING)
|
queue_entry.set_started_on_now()
|
|
|
def _write_lost_process_error_file(self):
|
error_file_path = os.path.join(self._working_directory(), 'job_failure')
|
_drone_manager.write_lines_to_file(error_file_path,
|
[_LOST_PROCESS_ERROR])
|
|
|
def _finish_task(self):
|
if not self.monitor:
|
return
|
|
self._write_job_finished()
|
|
if self.monitor.lost_process:
|
self._write_lost_process_error_file()
|
|
|
def _write_status_comment(self, comment):
|
_drone_manager.write_lines_to_file(
|
os.path.join(self._working_directory(), 'status.log'),
|
['INFO\t----\t----\t' + comment],
|
paired_with_process=self.monitor.get_process())
|
|
|
def _log_abort(self):
|
if not self.monitor or not self.monitor.has_process():
|
return
|
|
# build up sets of all the aborted_by and aborted_on values
|
aborted_by, aborted_on = set(), set()
|
for queue_entry in self.queue_entries:
|
if queue_entry.aborted_by:
|
aborted_by.add(queue_entry.aborted_by)
|
t = int(time.mktime(queue_entry.aborted_on.timetuple()))
|
aborted_on.add(t)
|
|
# extract some actual, unique aborted by value and write it out
|
# TODO(showard): this conditional is now obsolete, we just need to leave
|
# it in temporarily for backwards compatibility over upgrades. delete
|
# soon.
|
assert len(aborted_by) <= 1
|
if len(aborted_by) == 1:
|
aborted_by_value = aborted_by.pop()
|
aborted_on_value = max(aborted_on)
|
else:
|
aborted_by_value = 'autotest_system'
|
aborted_on_value = int(time.time())
|
|
self._write_keyval_after_job("aborted_by", aborted_by_value)
|
self._write_keyval_after_job("aborted_on", aborted_on_value)
|
|
aborted_on_string = str(datetime.datetime.fromtimestamp(
|
aborted_on_value))
|
self._write_status_comment('Job aborted by %s on %s' %
|
(aborted_by_value, aborted_on_string))
|
|
|
def abort(self):
|
super(AbstractQueueTask, self).abort()
|
self._log_abort()
|
self._finish_task()
|
|
|
def epilog(self):
|
super(AbstractQueueTask, self).epilog()
|
self._finish_task()
|
|
|
class QueueTask(AbstractQueueTask):
|
def __init__(self, queue_entries):
|
super(QueueTask, self).__init__(queue_entries)
|
self._set_ids(queue_entries=queue_entries)
|
self._enable_ssp_container = (
|
global_config.global_config.get_config_value(
|
'AUTOSERV', 'enable_ssp_container', type=bool,
|
default=True))
|
|
|
def prolog(self):
|
self._check_queue_entry_statuses(
|
self.queue_entries,
|
allowed_hqe_statuses=(models.HostQueueEntry.Status.STARTING,
|
models.HostQueueEntry.Status.RUNNING),
|
allowed_host_statuses=(models.Host.Status.PENDING,
|
models.Host.Status.RUNNING))
|
|
super(QueueTask, self).prolog()
|
|
for queue_entry in self.queue_entries:
|
self._write_host_keyvals(queue_entry.host)
|
queue_entry.host.set_status(models.Host.Status.RUNNING)
|
queue_entry.host.update_field('dirty', 1)
|
|
|
def _finish_task(self):
|
super(QueueTask, self)._finish_task()
|
|
for queue_entry in self.queue_entries:
|
queue_entry.set_status(models.HostQueueEntry.Status.GATHERING)
|
queue_entry.host.set_status(models.Host.Status.RUNNING)
|
|
|
def _command_line(self):
|
invocation = super(QueueTask, self)._command_line()
|
# Check if server-side packaging is needed.
|
if (self._enable_ssp_container and
|
self.job.control_type == control_data.CONTROL_TYPE.SERVER and
|
self.job.require_ssp != False):
|
invocation += ['--require-ssp']
|
keyval_dict = self.job.keyval_dict()
|
test_source_build = keyval_dict.get('test_source_build', None)
|
if test_source_build:
|
invocation += ['--test_source_build', test_source_build]
|
if self.job.parent_job_id:
|
invocation += ['--parent_job_id', str(self.job.parent_job_id)]
|
return invocation + ['--verify_job_repo_url']
|
|
|
class HostlessQueueTask(AbstractQueueTask):
|
def __init__(self, queue_entry):
|
super(HostlessQueueTask, self).__init__([queue_entry])
|
self.queue_entry_ids = [queue_entry.id]
|
|
|
def prolog(self):
|
super(HostlessQueueTask, self).prolog()
|
|
|
def _finish_task(self):
|
super(HostlessQueueTask, self)._finish_task()
|
|
# When a job is added to database, its initial status is always
|
# Starting. In a scheduler tick, scheduler finds all jobs in Starting
|
# status, check if any of them can be started. If scheduler hits some
|
# limit, e.g., max_hostless_jobs_per_drone, scheduler will
|
# leave these jobs in Starting status. Otherwise, the jobs'
|
# status will be changed to Running, and an autoserv process
|
# will be started in drone for each of these jobs.
|
# If the entry is still in status Starting, the process has not started
|
# yet. Therefore, there is no need to parse and collect log. Without
|
# this check, exception will be raised by scheduler as execution_subdir
|
# for this queue entry does not have a value yet.
|
hqe = self.queue_entries[0]
|
if hqe.status != models.HostQueueEntry.Status.STARTING:
|
hqe.set_status(models.HostQueueEntry.Status.PARSING)
|
|
|
if __name__ == '__main__':
|
main()
|
