# pylint: disable=missing-docstring """ This is the module for everything related to the AgentTask. The AgentTask imposes an interface through which the scheduler can monitor a processes; Examples of such processes include Verify, Cleanup and the Queue Tasks that run the tests. The scheduler itself only understands Agents. Agents: The Agent is the bridge between the scheduler and the AgentTask. The schedulers tick has a method called handle_agents, which calls the tick of each agent in the Dispatchers queue. This leads to the Agent polling its AgentTask. The scheduler will keep polling a task through the associated Agent till the Agent is removed from the dispatcher. At a high level: agents finished = tasks done agent polls till finished task polls till done task sets done agent is removed from dispatcher AgentTasks: Basic AgentTasks are created when an hqe changes state. Examples of these are the QueueTask, which is created when a hqe goes into the Starting state and the FinalReparseTask, which is created when the hqe goes into parsing. SpecialAgentTasks: Unlike AgentTasks, SpecialAgentTasks are only created when a row is inserted in the afe_special_tasks table. All PrejobTasks are SpecialAgentTasks. Monitor_db.get_agent_task_for_special_task/get_agent_task_for_queue_entry maps an AgentTask to an Agent, which the scheduler understands. From this point onward, the scheduler manages the task through the Agents interface,as follows: At a high level: task poll start prolog tick till we get an exit code finished(exit==0) done=True epilog cleanup set is_active, is_complete, success (checked in scheduler) The first special task for an HQE is usually Reset. -poll: The first poll will start the task, polls thereafter will call the tasks tick method. A started task will have the started bit set. - start: Call prolog, run the process and set the start bit. - prolog: Usually where one puts any model state changes that happen before the actual task. Different per Task. Examples of things that might happen in a prolog: - state of Host, HQE (to something like Resetting) - delete any unwanted queued special tasks - register a pidfile - set the is_active bit on the special task - run: - create a PidfileRunMonitor - pass the autoserv command, working directory etc to drone manager. This will start the actual autoserv process. - set the start bit: so subsequent polls do not 'start' again - tick: For as long as a started tasks done bit is not set, a poll will lead to a tick. The tick monitors the pid file of the autoserv process running on the drone through the PidfileRunMonitor created in prolog. If the autoserv process has finished we call finished with true/false depending on autoserv exit code. - finished: sets the done and success values, then calls epilog. The done bit is important because the Agent polls this bit to measure the success or failure of its task. - epilog: Is generally where we set status of the Host/HQE again, requeue any other task that needs to run after this one and perform cleanup. Just like the prolog, this step is different per task. - cleanup: Sets the is_active and is_complete and success states on the tasks model. Also uses the drone_manager to: unregister the pidfile copy results of the task (Note this is not to be confused with the special task called cleanup). The actions we take in the epilog are based on the success/failure of the autoserv process set in cleanup, eg: if reset failed we will enqueue a repair, but if all is well the epilog will just return. Prejob task epilogs also have an on_pending method that change the status of the HQE to pending/starting, which gets picked up in the scheduler. By this point the is_done flag is set, which results in the Agent noticing that the task has finished and unregistering it from the dispatcher.Class hierarchy: AgentTask |--->SpecialAgentTask (prejob_task.py) |--->RepairTask |--->PreJobTask |--->Verify, Cleanup, Reset, Provision |--->AbstractQueueTask (monitor_db.py) |--->QueueTask |--->HostlessQueueTask |--->PostJobTask (postjob_task.py) |--->GatherLogsTask |--->SelfThrottledPostJobTask |--->FinalReparseTask """ import logging import os import time import urllib import common 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 drone_manager from autotest_lib.scheduler import email_manager from autotest_lib.scheduler import pidfile_monitor from autotest_lib.scheduler import rdb_lib from autotest_lib.scheduler import scheduler_lib from autotest_lib.scheduler import scheduler_models from autotest_lib.server import autoserv_utils from autotest_lib.server import system_utils try: from chromite.lib import metrics except ImportError: metrics = utils.metrics_mock CONFIG = global_config.global_config AUTOSERV_NICE_LEVEL = 10 ENABLE_DRONE_IN_RESTRICTED_SUBNET = CONFIG.get_config_value( 'CROS', 'enable_drone_in_restricted_subnet', type=bool, default=False) class AgentTask(object): class _NullMonitor(object): pidfile_id = None def has_process(self): return True def __init__(self, log_file_name=None): """ @param log_file_name: (optional) name of file to log command output to """ self._drone_manager = drone_manager.instance() self.done = False self.started = False self.success = None self.aborted = False self.monitor = None self.queue_entry_ids = [] self.host_ids = [] # A map between host id and hostname. self.hostnames = {} self._log_file_name = log_file_name def _set_ids(self, host=None, queue_entries=None): if queue_entries and queue_entries != [None]: self.host_ids = [] self.queue_entry_ids = [] self.hostnames = {} for entry in queue_entries: if entry.host is not None: self.host_ids.append(entry.host.id) self.queue_entry_ids.append(entry.id) self.hostnames[entry.host.id] = entry.host.hostname else: logging.debug( 'No host is found for host_queue_entry_id: %r', entry.id) raise scheduler_lib.NoHostIdError( 'Failed to schedule a job whose ' 'host_queue_entry_id=%r due to no host_id.' % entry.id) else: assert host self.host_ids = [host.id] self.hostnames = {host.id: host.hostname} def poll(self): if not self.started: self.start() if not self.done: self.tick() def tick(self): assert self.monitor exit_code = self.monitor.exit_code() if exit_code is None: return success = (exit_code == 0) self.finished(success) def is_done(self): return self.done def finished(self, success): if self.done: assert self.started return self.started = True self.done = True self.success = success self.epilog() def prolog(self): """ To be overridden. """ assert not self.monitor self.register_necessary_pidfiles() def _log_file(self): if not self._log_file_name: return None return os.path.join(self._working_directory(), self._log_file_name) def cleanup(self): log_file = self._log_file() if self.monitor and log_file: self.monitor.try_copy_to_results_repository(log_file) def epilog(self): """ To be overridden. """ self.cleanup() logging.info("%s finished with success=%s", type(self).__name__, self.success) def start(self): if not self.started: self.prolog() self.run() self.started = True def abort(self): if self.monitor: self.monitor.kill() self.done = True self.aborted = True self.cleanup() def _get_consistent_execution_path(self, execution_entries): first_execution_path = execution_entries[0].execution_path() for execution_entry in execution_entries[1:]: assert execution_entry.execution_path() == first_execution_path, ( '%s (%s) != %s (%s)' % (execution_entry.execution_path(), execution_entry, first_execution_path, execution_entries[0])) return first_execution_path def _copy_results(self, execution_entries, use_monitor=None): """ @param execution_entries: list of objects with execution_path() method """ if use_monitor is not None and not use_monitor.has_process(): return assert len(execution_entries) > 0 if use_monitor is None: assert self.monitor use_monitor = self.monitor assert use_monitor.has_process() execution_path = self._get_consistent_execution_path(execution_entries) results_path = execution_path + '/' use_monitor.try_copy_to_results_repository(results_path) def _parse_results(self, queue_entries): for queue_entry in queue_entries: queue_entry.set_status(models.HostQueueEntry.Status.PARSING) def _command_line(self): """ Return the command line to run. Must be overridden. """ raise NotImplementedError @property def num_processes(self): """ Return the number of processes forked by this AgentTask's process. It may only be approximate. To be overridden if necessary. """ return 1 def _paired_with_monitor(self): """ If this AgentTask's process must run on the same machine as some previous process, this method should be overridden to return a PidfileRunMonitor for that process. """ return self._NullMonitor() @property def owner_username(self): """ Return login of user responsible for this task. May be None. Must be overridden. """ raise NotImplementedError def _working_directory(self): """ Return the directory where this AgentTask's process executes. Must be overridden. """ raise NotImplementedError def _pidfile_name(self): """ Return the name of the pidfile this AgentTask's process uses. To be overridden if necessary. """ return drone_manager.AUTOSERV_PID_FILE def _check_paired_results_exist(self): if not self._paired_with_monitor().has_process(): metrics.Counter( 'chromeos/autotest/errors/scheduler/no_paired_results' ).increment() self.finished(False) return False return True def _create_monitor(self): assert not self.monitor self.monitor = pidfile_monitor.PidfileRunMonitor() def run(self): if not self._check_paired_results_exist(): return self._create_monitor() self.monitor.run( self._command_line(), self._working_directory(), num_processes=self.num_processes, nice_level=AUTOSERV_NICE_LEVEL, log_file=self._log_file(), pidfile_name=self._pidfile_name(), paired_with_pidfile=self._paired_with_monitor().pidfile_id, username=self.owner_username, drone_hostnames_allowed=self.get_drone_hostnames_allowed()) def get_drone_hostnames_allowed( self, restricted_subnets=utils.RESTRICTED_SUBNETS, enable_drone_in_subnet=ENABLE_DRONE_IN_RESTRICTED_SUBNET): filtered_drones = None has_unrestricted_host = False if (self.hostnames and restricted_subnets and enable_drone_in_subnet): for hostname in self.hostnames.values(): subnet = utils.get_restricted_subnet(hostname, restricted_subnets) # Return an empty set if the list of hosts exists both in # restricted and unrestricted subnet. No drone can work in such # case. if ((not subnet and filtered_drones is not None) or (subnet and has_unrestricted_host)): logging.error('The test has some DUT in restricted subnet, ' 'but some in unrestricted subnet. Therefore, ' 'no drone is available to run the test.') return set() if not subnet: has_unrestricted_host = True continue server_ip_map=system_utils.DroneCache.get_drone_ip_map() filtered_drones_for_host = set( utils.get_servers_in_same_subnet( subnet[0], subnet[1], server_ip_map=server_ip_map)) logging.info('DUT %s is in restricted subnet, drone can only ' 'be chosen from %s', hostname, filtered_drones_for_host) if filtered_drones is None: filtered_drones = filtered_drones_for_host else: filtered_drones = set.intersection( filtered_drones, filtered_drones_for_host) # If filtered_drones is an empty set, that means no drone is # allowed to run the task. This is different fron None, which # means all drones are allowed. if filtered_drones == set(): logging.error('DUT(s) is in restricted subnet, but no ' 'drone is available to run the test.') return filtered_drones # If host is not in restricted subnet, use the unrestricted drones only. if (filtered_drones is None and restricted_subnets and enable_drone_in_subnet): filtered_drones = set( system_utils.DroneCache.get_unrestricted_drones( restricted_subnets=restricted_subnets)) if not models.DroneSet.drone_sets_enabled(): return filtered_drones hqes = models.HostQueueEntry.objects.filter(id__in=self.queue_entry_ids) if not hqes: # Only special tasks could be missing host queue entries assert isinstance(self, SpecialAgentTask) return self._user_or_global_default_drone_set( self.task, self.task.requested_by) job_ids = hqes.values_list('job', flat=True).distinct() assert job_ids.count() == 1, ("AgentTask's queue entries " "span multiple jobs") job = models.Job.objects.get(id=job_ids[0]) drone_set = job.drone_set if not drone_set: return self._user_or_global_default_drone_set(job, job.user()) if filtered_drones: return set.intersection(filtered_drones, drone_set.get_drone_hostnames()) else: return drone_set.get_drone_hostnames() def _user_or_global_default_drone_set(self, obj_with_owner, user): """ Returns the user's default drone set, if present. Otherwise, returns the global default drone set. """ default_hostnames = models.DroneSet.get_default().get_drone_hostnames() if not user: logging.warning('%s had no owner; using default drone set', obj_with_owner) return default_hostnames if not user.drone_set: logging.warning('User %s has no default drone set, using global ' 'default', user.login) return default_hostnames return user.drone_set.get_drone_hostnames() def register_necessary_pidfiles(self): pidfile_id = self._drone_manager.get_pidfile_id_from( self._working_directory(), self._pidfile_name()) self._drone_manager.register_pidfile(pidfile_id) paired_pidfile_id = self._paired_with_monitor().pidfile_id if paired_pidfile_id: self._drone_manager.register_pidfile(paired_pidfile_id) def recover(self): if not self._check_paired_results_exist(): return self._create_monitor() self.monitor.attach_to_existing_process( self._working_directory(), pidfile_name=self._pidfile_name(), num_processes=self.num_processes) if not self.monitor.has_process(): # no process to recover; wait to be started normally self.monitor = None return self.started = True logging.info('Recovering process %s for %s at %s', self.monitor.get_process(), type(self).__name__, self._working_directory()) def _check_queue_entry_statuses(self, queue_entries, allowed_hqe_statuses, allowed_host_statuses=None): class_name = self.__class__.__name__ for entry in queue_entries: if entry.status not in allowed_hqe_statuses: # In the orignal code, here we raise an exception. In an # effort to prevent downtime we will instead abort the job and # send out an email notifying us this has occured. error_message = ('%s attempting to start entry with invalid ' 'status %s: %s. Aborting Job: %s.' % (class_name, entry.status, entry, entry.job)) logging.error(error_message) email_manager.manager.enqueue_notify_email( 'Job Aborted - Invalid Host Queue Entry Status', error_message) entry.job.request_abort() invalid_host_status = ( allowed_host_statuses is not None and entry.host.status not in allowed_host_statuses) if invalid_host_status: # In the orignal code, here we raise an exception. In an # effort to prevent downtime we will instead abort the job and # send out an email notifying us this has occured. error_message = ('%s attempting to start on queue entry with ' 'invalid host status %s: %s. Aborting Job: %s' % (class_name, entry.host.status, entry, entry.job)) logging.error(error_message) email_manager.manager.enqueue_notify_email( 'Job Aborted - Invalid Host Status', error_message) entry.job.request_abort() class TaskWithJobKeyvals(object): """AgentTask mixin providing functionality to help with job keyval files.""" _KEYVAL_FILE = 'keyval' def _format_keyval(self, key, value): return '%s=%s' % (key, value) def _keyval_path(self): """Subclasses must override this""" raise NotImplementedError def _write_keyval_after_job(self, field, value): assert self.monitor if not self.monitor.has_process(): return self._drone_manager.write_lines_to_file( self._keyval_path(), [self._format_keyval(field, value)], paired_with_process=self.monitor.get_process()) def _job_queued_keyval(self, job): return 'job_queued', int(time.mktime(job.created_on.timetuple())) def _write_job_finished(self): self._write_keyval_after_job("job_finished", int(time.time())) def _write_keyvals_before_job_helper(self, keyval_dict, keyval_path): keyval_contents = '\n'.join(self._format_keyval(key, value) for key, value in keyval_dict.iteritems()) # always end with a newline to allow additional keyvals to be written keyval_contents += '\n' self._drone_manager.attach_file_to_execution(self._working_directory(), keyval_contents, file_path=keyval_path) def _write_keyvals_before_job(self, keyval_dict): self._write_keyvals_before_job_helper(keyval_dict, self._keyval_path()) def _write_host_keyvals(self, host): keyval_path = os.path.join(self._working_directory(), 'host_keyvals', host.hostname) platform, all_labels = host.platform_and_labels() all_labels = [ urllib.quote(label) for label in all_labels ] keyval_dict = dict(platform=platform, labels=','.join(all_labels)) self._write_keyvals_before_job_helper(keyval_dict, keyval_path) class SpecialAgentTask(AgentTask, TaskWithJobKeyvals): """ Subclass for AgentTasks that correspond to a SpecialTask entry in the DB. """ TASK_TYPE = None host = None queue_entry = None _COUNT_METRIC = 'chromeos/autotest/scheduler/special_task_count' _DUT_METRIC = 'chromeos/autotest/scheduler/special_task_by_dut' _DURATION_METRIC = 'chromeos/autotest/scheduler/special_task_durations' def __init__(self, task, extra_command_args): super(SpecialAgentTask, self).__init__() assert self.TASK_TYPE is not None, 'self.TASK_TYPE must be overridden' self.host = rdb_lib.get_hosts([task.host.id])[0] self.host.dbg_str = 'Task: %s' % str(task) self.queue_entry = None if task.queue_entry: self.queue_entry = scheduler_models.HostQueueEntry( id=task.queue_entry.id) self.host.dbg_str += self.queue_entry.get_dbg_str() # This is of type SpecialTask (as defined in frontend/afe/models.py) self.task = task self._extra_command_args = extra_command_args self.host.metadata = self.get_metadata() self._milestone = '' def get_metadata(self): """Get a dictionary that contains task information. The return value is a dictionary that includes task information like id, name and related job information. The value will be stored in metadata database. @return: A dictionary containing the task id, name and related job id. If some attributes are failed to be accessed, an empty dictionary will be returned, and error will be logged. """ try: metadata = {'task_id':self.task.id, 'task_name':self.task.task, 'hostname':self.task.host.hostname} if self.task.queue_entry: job = self.task.queue_entry.job metadata.update( scheduler_models.get_job_metadata(job)) return metadata except AttributeError as e: logging.error('Task has missing attribute: %s', e) return {} def _keyval_path(self): return os.path.join(self._working_directory(), self._KEYVAL_FILE) def _command_line(self): return autoserv_utils.autoserv_run_job_command( autoserv_utils.autoserv_directory, self.host.hostname, results_directory=drone_manager.WORKING_DIRECTORY, extra_args=self._extra_command_args, queue_entry=self.queue_entry, in_lab=True, ) def _working_directory(self): return self.task.execution_path() @property def owner_username(self): if self.task.requested_by: return self.task.requested_by.login return None def prolog(self): super(SpecialAgentTask, self).prolog() self.task.activate() self._write_host_keyvals(self.host) def _fail_queue_entry(self): assert self.queue_entry if self.queue_entry.meta_host: return # don't fail metahost entries, they'll be reassigned self.queue_entry.update_from_database() if self.queue_entry.status != models.HostQueueEntry.Status.QUEUED: return # entry has been aborted self._actually_fail_queue_entry() def epilog(self): super(SpecialAgentTask, self).epilog() self._emit_special_task_status_metric() def _emit_special_task_status_metric(self): """Increments an accumulator associated with this special task.""" fields = {'type': self.TASK_TYPE, 'success': bool(self.success), 'board': str(self.host.board), 'milestone': self._milestone} metrics.Counter(self._COUNT_METRIC).increment( fields=fields) if (self.task.time_finished and self.task.time_started): duration = (self.task.time_finished - self.task.time_started).total_seconds() metrics.SecondsDistribution(self._DURATION_METRIC).add( duration, fields=fields) dut_fields = { 'type': self.TASK_TYPE, 'success': bool(self.success), 'board': str(self.host.board), 'dut_host_name': self.host.hostname } metrics.Counter(self._DUT_METRIC).increment(fields=dut_fields) # TODO(milleral): http://crbug.com/268607 # All this used to be a part of _fail_queue_entry. The # exact semantics of when one should and should not be failing a queue # entry need to be worked out, because provisioning has placed us in a # case where we want to fail a queue entry that could be requeued, # which makes us fail the two above if statements, and thus # _fail_queue_entry() would exit early and have no effect. # What's left here with _actually_fail_queue_entry is a hack to be able to # bypass the checks and unconditionally execute the code. def _actually_fail_queue_entry(self): self.queue_entry.set_execution_subdir() queued_key, queued_time = self._job_queued_keyval( self.queue_entry.job) self._write_keyval_after_job(queued_key, queued_time) self._write_job_finished() # copy results logs into the normal place for job results self.monitor.try_copy_results_on_drone( source_path=self._working_directory() + '/', destination_path=self.queue_entry.execution_path() + '/') pidfile_id = self._drone_manager.get_pidfile_id_from( self.queue_entry.execution_path(), pidfile_name=drone_manager.AUTOSERV_PID_FILE) self._drone_manager.register_pidfile(pidfile_id) # TODO(ayatane): This should obey self.queue_entry.job.parse_failed_repair # But nothing sets self.queue_entry.job.parse_failed_repair? # Check Git blame self._parse_results([self.queue_entry]) # Also fail all other special tasks that have not yet run for this HQE pending_tasks = models.SpecialTask.objects.filter( queue_entry__id=self.queue_entry.id, is_complete=0) for task in pending_tasks: task.finish(False) def cleanup(self): super(SpecialAgentTask, self).cleanup() # We will consider an aborted task to be "Failed" self.task.finish(bool(self.success)) if self.monitor: if self.monitor.has_process(): self._copy_results([self.task]) if self.monitor.pidfile_id is not None: self._drone_manager.unregister_pidfile(self.monitor.pidfile_id) def remove_special_tasks(self, special_task_to_remove, keep_last_one=False): """Remove a type of special task in all tasks, keep last one if needed. @param special_task_to_remove: type of special task to be removed, e.g., models.SpecialTask.Task.VERIFY. @param keep_last_one: True to keep the last special task if its type is the same as of special_task_to_remove. """ queued_special_tasks = models.SpecialTask.objects.filter( host__id=self.host.id, task=special_task_to_remove, is_active=False, is_complete=False, queue_entry=None) if keep_last_one: queued_special_tasks = queued_special_tasks.exclude(id=self.task.id) queued_special_tasks.delete() def _generate_autoserv_label_args(self, task): """ @param task: An instance of afe model's SpecialTask. @returns: The list of arguments to pass to autoserv to tell it what the labels of a job are. """ labels = {x.name for x in task.queue_entry.job.labels} return ['--job-labels', ','.join(labels)]