#pylint: disable-msg=C0111
|
|
"""
|
Postjob task.
|
|
Postjob tasks are responsible for setting the final status of the HQE
|
and Host, and scheduling additional special agents such as cleanup,
|
if necessary.
|
"""
|
|
import os
|
|
from autotest_lib.client.common_lib import utils
|
from autotest_lib.frontend.afe import models, model_attributes
|
from autotest_lib.scheduler import agent_task, drones, drone_manager
|
from autotest_lib.scheduler import email_manager, pidfile_monitor
|
from autotest_lib.scheduler import scheduler_config
|
from autotest_lib.server import autoserv_utils
|
|
try:
|
from chromite.lib import metrics
|
except ImportError:
|
metrics = utils.metrics_mock
|
|
|
_parser_path = os.path.join(drones.AUTOTEST_INSTALL_DIR, 'tko', 'parse')
|
|
|
class PostJobTask(agent_task.AgentTask):
|
def __init__(self, queue_entries, log_file_name):
|
super(PostJobTask, self).__init__(log_file_name=log_file_name)
|
|
self.queue_entries = queue_entries
|
|
self._autoserv_monitor = pidfile_monitor.PidfileRunMonitor()
|
self._autoserv_monitor.attach_to_existing_process(
|
self._working_directory())
|
|
|
def _command_line(self):
|
# Do we need testing_mode?
|
return self._generate_command(
|
self._drone_manager.absolute_path(self._working_directory()))
|
|
|
def _generate_command(self, results_dir):
|
raise NotImplementedError('Subclasses must override this')
|
|
|
@property
|
def owner_username(self):
|
return self.queue_entries[0].job.owner
|
|
|
def _working_directory(self):
|
return self._get_consistent_execution_path(self.queue_entries)
|
|
|
def _paired_with_monitor(self):
|
return self._autoserv_monitor
|
|
|
def _job_was_aborted(self):
|
was_aborted = None
|
for queue_entry in self.queue_entries:
|
queue_entry.update_from_database()
|
if was_aborted is None: # first queue entry
|
was_aborted = bool(queue_entry.aborted)
|
elif was_aborted != bool(queue_entry.aborted): # subsequent entries
|
entries = ['%s (aborted: %s)' % (entry, entry.aborted)
|
for entry in self.queue_entries]
|
email_manager.manager.enqueue_notify_email(
|
'Inconsistent abort state',
|
'Queue entries have inconsistent abort state:\n' +
|
'\n'.join(entries))
|
# don't crash here, just assume true
|
return True
|
return was_aborted
|
|
|
def _final_status(self):
|
if self._job_was_aborted():
|
return models.HostQueueEntry.Status.ABORTED
|
|
# we'll use a PidfileRunMonitor to read the autoserv exit status
|
if self._autoserv_monitor.exit_code() == 0:
|
return models.HostQueueEntry.Status.COMPLETED
|
return models.HostQueueEntry.Status.FAILED
|
|
|
def _set_all_statuses(self, status):
|
for queue_entry in self.queue_entries:
|
queue_entry.set_status(status)
|
|
|
def abort(self):
|
# override AgentTask.abort() to avoid killing the process and ending
|
# the task. post-job tasks continue when the job is aborted.
|
pass
|
|
|
def _pidfile_label(self):
|
# '.autoserv_execute' -> 'autoserv'
|
return self._pidfile_name()[1:-len('_execute')]
|
|
|
class SelfThrottledPostJobTask(PostJobTask):
|
"""
|
PostJobTask that maintains its own process limit.
|
|
We throttle tasks like parsing because we don't want them to
|
hold up tests. At the same time we don't wish to build up load
|
that will take forever to parse.
|
"""
|
_num_running_processes = 0
|
# Last known limit of max processes, used to check whether
|
# max processes config has been changed.
|
_last_known_max_processes = 0
|
# Whether an email should be sent to notifiy process limit being hit.
|
_notification_on = True
|
# Once process limit is hit, an email will be sent.
|
# To prevent spams, do not send another email until
|
# it drops to lower than the following level.
|
REVIVE_NOTIFICATION_THRESHOLD = 0.80
|
|
@classmethod
|
def _gauge_metrics(cls):
|
"""Report to monarch the number of running processes."""
|
m = metrics.Gauge('chromeos/autotest/scheduler/postjob_tasks')
|
m.set(cls._num_running_processes, fields={'task_name': cls.__name__})
|
|
|
@classmethod
|
def _increment_running_processes(cls):
|
cls._num_running_processes += 1
|
cls._gauge_metrics()
|
|
|
@classmethod
|
def _decrement_running_processes(cls):
|
cls._num_running_processes -= 1
|
cls._gauge_metrics()
|
|
|
@classmethod
|
def _max_processes(cls):
|
raise NotImplementedError
|
|
|
@classmethod
|
def _can_run_new_process(cls):
|
return cls._num_running_processes < cls._max_processes()
|
|
|
def _process_started(self):
|
return bool(self.monitor)
|
|
|
def tick(self):
|
# override tick to keep trying to start until the process count goes
|
# down and we can, at which point we revert to default behavior
|
if self._process_started():
|
super(SelfThrottledPostJobTask, self).tick()
|
else:
|
self._try_starting_process()
|
|
|
def run(self):
|
# override run() to not actually run unless we can
|
self._try_starting_process()
|
|
|
@classmethod
|
def _notify_process_limit_hit(cls):
|
"""Send an email to notify that process limit is hit."""
|
if cls._notification_on:
|
subject = '%s: hitting max process limit.' % cls.__name__
|
message = ('Running processes/Max processes: %d/%d'
|
% (cls._num_running_processes, cls._max_processes()))
|
email_manager.manager.enqueue_notify_email(subject, message)
|
cls._notification_on = False
|
|
|
@classmethod
|
def _reset_notification_switch_if_necessary(cls):
|
"""Reset _notification_on if necessary.
|
|
Set _notification_on to True on the following cases:
|
1) If the limit of max processes configuration changes;
|
2) If _notification_on is False and the number of running processes
|
drops to lower than a level defined in REVIVE_NOTIFICATION_THRESHOLD.
|
|
"""
|
if cls._last_known_max_processes != cls._max_processes():
|
cls._notification_on = True
|
cls._last_known_max_processes = cls._max_processes()
|
return
|
percentage = float(cls._num_running_processes) / cls._max_processes()
|
if (not cls._notification_on and
|
percentage < cls.REVIVE_NOTIFICATION_THRESHOLD):
|
cls._notification_on = True
|
|
|
def _try_starting_process(self):
|
self._reset_notification_switch_if_necessary()
|
if not self._can_run_new_process():
|
self._notify_process_limit_hit()
|
return
|
|
# actually run the command
|
super(SelfThrottledPostJobTask, self).run()
|
if self._process_started():
|
self._increment_running_processes()
|
|
|
def finished(self, success):
|
super(SelfThrottledPostJobTask, self).finished(success)
|
if self._process_started():
|
self._decrement_running_processes()
|
|
|
class GatherLogsTask(PostJobTask):
|
"""
|
Task responsible for
|
* gathering uncollected logs (if Autoserv crashed hard or was killed)
|
* copying logs to the results repository
|
* spawning CleanupTasks for hosts, if necessary
|
* spawning a FinalReparseTask for the job
|
* setting the final status of the host, directly or through a cleanup
|
"""
|
def __init__(self, queue_entries, recover_run_monitor=None):
|
self._job = queue_entries[0].job
|
super(GatherLogsTask, self).__init__(
|
queue_entries, log_file_name='.collect_crashinfo.log')
|
self._set_ids(queue_entries=queue_entries)
|
|
|
# TODO: Refactor into autoserv_utils. crbug.com/243090
|
def _generate_command(self, results_dir):
|
host_list = ','.join(queue_entry.host.hostname
|
for queue_entry in self.queue_entries)
|
return [autoserv_utils.autoserv_path , '-p',
|
'--pidfile-label=%s' % self._pidfile_label(),
|
'--use-existing-results', '--collect-crashinfo',
|
'-m', host_list, '-r', results_dir]
|
|
|
@property
|
def num_processes(self):
|
return len(self.queue_entries)
|
|
|
def _pidfile_name(self):
|
return drone_manager.CRASHINFO_PID_FILE
|
|
|
def prolog(self):
|
self._check_queue_entry_statuses(
|
self.queue_entries,
|
allowed_hqe_statuses=(models.HostQueueEntry.Status.GATHERING,),
|
allowed_host_statuses=(models.Host.Status.RUNNING,))
|
|
super(GatherLogsTask, self).prolog()
|
|
|
def epilog(self):
|
super(GatherLogsTask, self).epilog()
|
self._parse_results(self.queue_entries)
|
final_success, num_tests_failed = self._get_monitor_info()
|
reset_after_failure = (
|
not self._job.run_reset and (
|
not final_success or num_tests_failed > 0))
|
self._reboot_hosts(final_success, num_tests_failed, reset_after_failure)
|
if reset_after_failure:
|
m = metrics.Counter('chromeos/autotest/scheduler/postjob_tasks/'
|
'reset_after_failure')
|
m.increment(fields={'autoserv_process_success': final_success,
|
'num_tests_failed': num_tests_failed > 0})
|
self._reset_after_failure()
|
|
|
def _get_monitor_info(self):
|
"""Read monitor info from pidfile.
|
|
@returns: a tuple including
|
final_success: whether the monitor is successfully finished;
|
num_tests_failed: how many failed tests in the process.
|
"""
|
if self._autoserv_monitor.has_process():
|
final_success = (self._final_status() ==
|
models.HostQueueEntry.Status.COMPLETED)
|
num_tests_failed = self._autoserv_monitor.num_tests_failed()
|
else:
|
final_success = False
|
num_tests_failed = 0
|
|
return final_success, num_tests_failed
|
|
|
def _reboot_hosts(self, final_success, num_tests_failed,
|
reset_after_failure):
|
"""Reboot hosts by scheduling a CLEANUP task on host if needed.
|
|
@param final_success: whether the monitor successfully exits.
|
@param num_tests_failed: how many failed tests in total.
|
@param reset_after_failure: whether to schedule RESET task later.
|
"""
|
reboot_after = self._job.reboot_after
|
do_reboot = (
|
# always reboot after aborted jobs
|
self._final_status() == models.HostQueueEntry.Status.ABORTED
|
or reboot_after == model_attributes.RebootAfter.ALWAYS
|
or (reboot_after == model_attributes.RebootAfter.IF_ALL_TESTS_PASSED
|
and final_success and num_tests_failed == 0)
|
or (num_tests_failed > 0 and not reset_after_failure))
|
|
for queue_entry in self.queue_entries:
|
if do_reboot:
|
# don't pass the queue entry to the CleanupTask. if the cleanup
|
# fails, the job doesn't care -- it's over.
|
models.SpecialTask.objects.create(
|
host=models.Host.objects.get(id=queue_entry.host.id),
|
task=models.SpecialTask.Task.CLEANUP,
|
requested_by=self._job.owner_model())
|
else:
|
queue_entry.host.set_status(models.Host.Status.READY)
|
|
|
def _reset_after_failure(self):
|
"""Queue a RESET job for the host if job fails.
|
|
The current hqe entry is not passed into the RESET job.
|
"""
|
for queue_entry in self.queue_entries:
|
models.SpecialTask.objects.create(
|
host=models.Host.objects.get(id=queue_entry.host.id),
|
task=models.SpecialTask.Task.RESET,
|
requested_by=self._job.owner_model())
|
|
|
def run(self):
|
autoserv_exit_code = self._autoserv_monitor.exit_code()
|
# only run if Autoserv exited due to some signal. if we have no exit
|
# code, assume something bad (and signal-like) happened.
|
if autoserv_exit_code is None or os.WIFSIGNALED(autoserv_exit_code):
|
super(GatherLogsTask, self).run()
|
else:
|
self.finished(True)
|
|
|
class FinalReparseTask(SelfThrottledPostJobTask):
|
def __init__(self, queue_entries):
|
super(FinalReparseTask, self).__init__(queue_entries,
|
log_file_name='.parse.log')
|
# don't use _set_ids, since we don't want to set the host_ids
|
self.queue_entry_ids = [entry.id for entry in queue_entries]
|
|
|
def _generate_command(self, results_dir):
|
return [_parser_path, '--detach', '--write-pidfile',
|
'--record-duration', '--suite-report', '-l', '2', '-r', '-o',
|
results_dir]
|
|
|
@property
|
def num_processes(self):
|
return 0 # don't include parser processes in accounting
|
|
|
def _pidfile_name(self):
|
return drone_manager.PARSER_PID_FILE
|
|
|
@classmethod
|
def _max_processes(cls):
|
return scheduler_config.config.max_parse_processes
|
|
|
def prolog(self):
|
self._check_queue_entry_statuses(
|
self.queue_entries,
|
allowed_hqe_statuses=(models.HostQueueEntry.Status.PARSING,))
|
|
super(FinalReparseTask, self).prolog()
|
|
|
def epilog(self):
|
super(FinalReparseTask, self).epilog()
|
self._set_all_statuses(self._final_status())
|
