scheduler/drone_manager.py

import heapq
import os
import logging

import common
from autotest_lib.client.common_lib import error
from autotest_lib.client.common_lib import global_config
from autotest_lib.client.common_lib import utils
from autotest_lib.scheduler import drone_task_queue
from autotest_lib.scheduler import drone_utility
from autotest_lib.scheduler import drones
from autotest_lib.scheduler import scheduler_config
from autotest_lib.scheduler import thread_lib

try:
    from chromite.lib import metrics
except ImportError:
    metrics = utils.metrics_mock


# results on drones will be placed under the drone_installation_directory in a
# directory with this name
_DRONE_RESULTS_DIR_SUFFIX = 'results'

WORKING_DIRECTORY = object() # see execute_command()


AUTOSERV_PID_FILE = '.autoserv_execute'
CRASHINFO_PID_FILE = '.collect_crashinfo_execute'
PARSER_PID_FILE = '.parser_execute'
ARCHIVER_PID_FILE = '.archiver_execute'

ALL_PIDFILE_NAMES = (AUTOSERV_PID_FILE, CRASHINFO_PID_FILE, PARSER_PID_FILE,
                     ARCHIVER_PID_FILE)

_THREADED_DRONE_MANAGER = global_config.global_config.get_config_value(
        scheduler_config.CONFIG_SECTION, 'threaded_drone_manager',
        type=bool, default=True)

HOSTS_JOB_SUBDIR = 'hosts/'
PARSE_LOG = '.parse.log'
ENABLE_ARCHIVING =  global_config.global_config.get_config_value(
        scheduler_config.CONFIG_SECTION, 'enable_archiving', type=bool)


class DroneManagerError(Exception):
    pass


class CustomEquals(object):
    def _id(self):
        raise NotImplementedError


    def __eq__(self, other):
        if not isinstance(other, type(self)):
            return NotImplemented
        return self._id() == other._id()


    def __ne__(self, other):
        return not self == other


    def __hash__(self):
        return hash(self._id())


class Process(CustomEquals):
    def __init__(self, hostname, pid, ppid=None):
        self.hostname = hostname
        self.pid = pid
        self.ppid = ppid

    def _id(self):
        return (self.hostname, self.pid)


    def __str__(self):
        return '%s/%s' % (self.hostname, self.pid)


    def __repr__(self):
        return super(Process, self).__repr__() + '<%s>' % self


class PidfileId(CustomEquals):
    def __init__(self, path):
        self.path = path


    def _id(self):
        return self.path


    def __str__(self):
        return str(self.path)


class _PidfileInfo(object):
    age = 0
    num_processes = None


class PidfileContents(object):
    process = None
    exit_status = None
    num_tests_failed = None

    def is_invalid(self):
        return False


    def is_running(self):
        return self.process and not self.exit_status


class InvalidPidfile(object):
    process = None
    exit_status = None
    num_tests_failed = None


    def __init__(self, error):
        self.error = error


    def is_invalid(self):
        return True


    def is_running(self):
        return False


    def __str__(self):
        return self.error


class _DroneHeapWrapper(object):
    """Wrapper to compare drones based on used_capacity().

    These objects can be used to keep a heap of drones by capacity.
    """
    def __init__(self, drone):
        self.drone = drone


    def __cmp__(self, other):
        assert isinstance(other, _DroneHeapWrapper)
        return cmp(self.drone.used_capacity(), other.drone.used_capacity())


class DroneManager(object):
    """
    This class acts as an interface from the scheduler to drones, whether it be
    only a single "drone" for localhost or multiple remote drones.

    All paths going into and out of this class are relative to the full results
    directory, except for those returns by absolute_path().
    """


    # Minimum time to wait before next email
    # about a drone hitting process limit is sent.
    NOTIFY_INTERVAL = 60 * 60 * 24 # one day
    _STATS_KEY = 'drone_manager'



    def __init__(self):
        # absolute path of base results dir
        self._results_dir = None
        # holds Process objects
        self._process_set = set()
        # holds the list of all processes running on all drones
        self._all_processes = {}
        # maps PidfileId to PidfileContents
        self._pidfiles = {}
        # same as _pidfiles
        self._pidfiles_second_read = {}
        # maps PidfileId to _PidfileInfo
        self._registered_pidfile_info = {}
        # used to generate unique temporary paths
        self._temporary_path_counter = 0
        # maps hostname to Drone object
        self._drones = {}
        self._results_drone = None
        # maps results dir to dict mapping file path to contents
        self._attached_files = {}
        # heapq of _DroneHeapWrappers
        self._drone_queue = []
        # A threaded task queue used to refresh drones asynchronously.
        if _THREADED_DRONE_MANAGER:
            self._refresh_task_queue = thread_lib.ThreadedTaskQueue(
                    name='%s.refresh_queue' % self._STATS_KEY)
        else:
            self._refresh_task_queue = drone_task_queue.DroneTaskQueue()


    def initialize(self, base_results_dir, drone_hostnames,
                   results_repository_hostname):
        self._results_dir = base_results_dir

        for hostname in drone_hostnames:
            self._add_drone(hostname)

        if not self._drones:
            # all drones failed to initialize
            raise DroneManagerError('No valid drones found')

        self.refresh_drone_configs()

        logging.info('Using results repository on %s',
                     results_repository_hostname)
        self._results_drone = drones.get_drone(results_repository_hostname)
        results_installation_dir = global_config.global_config.get_config_value(
                scheduler_config.CONFIG_SECTION,
                'results_host_installation_directory', default=None)
        if results_installation_dir:
            self._results_drone.set_autotest_install_dir(
                    results_installation_dir)
        # don't initialize() the results drone - we don't want to clear out any
        # directories and we don't need to kill any processes


    def reinitialize_drones(self):
        for drone in self.get_drones():
            with metrics.SecondsTimer(
                    'chromeos/autotest/drone_manager/'
                    'reinitialize_drones_duration',
                    fields={'drone': drone.hostname}):
                drone.call('initialize', self._results_dir)


    def shutdown(self):
        for drone in self.get_drones():
            drone.shutdown()


    def _get_max_pidfile_refreshes(self):
        """
        Normally refresh() is called on every monitor_db.Dispatcher.tick().

        @returns: The number of refresh() calls before we forget a pidfile.
        """
        pidfile_timeout = global_config.global_config.get_config_value(
                scheduler_config.CONFIG_SECTION, 'max_pidfile_refreshes',
                type=int, default=2000)
        return pidfile_timeout


    def _add_drone(self, hostname):
        """
        Add drone.

        Catches AutoservRunError if the drone fails initialization and does not
        add it to the list of usable drones.

        @param hostname: Hostname of the drone we are trying to add.
        """
        logging.info('Adding drone %s' % hostname)
        drone = drones.get_drone(hostname)
        if drone:
            try:
                drone.call('initialize', self.absolute_path(''))
            except error.AutoservRunError as e:
                logging.error('Failed to initialize drone %s with error: %s',
                              hostname, e)
                return
            self._drones[drone.hostname] = drone


    def _remove_drone(self, hostname):
        self._drones.pop(hostname, None)


    def refresh_drone_configs(self):
        """
        Reread global config options for all drones.
        """
        # Import server_manager_utils is delayed rather than at the beginning of
        # this module. The reason is that test_that imports drone_manager when
        # importing autoserv_utils. The import is done before test_that setup
        # django (test_that only setup django in setup_local_afe, since it's
        # not needed when test_that runs the test in a lab duts through :lab:
        # option. Therefore, if server_manager_utils is imported at the
        # beginning of this module, test_that will fail since django is not
        # setup yet.
        from autotest_lib.site_utils import server_manager_utils
        config = global_config.global_config
        section = scheduler_config.CONFIG_SECTION
        config.parse_config_file()
        for hostname, drone in self._drones.iteritems():
            if server_manager_utils.use_server_db():
                server = server_manager_utils.get_servers(hostname=hostname)[0]
                attributes = dict([(a.attribute, a.value)
                                   for a in server.attributes.all()])
                drone.enabled = (
                        int(attributes.get('disabled', 0)) == 0)
                drone.max_processes = int(
                        attributes.get(
                            'max_processes',
                            scheduler_config.config.max_processes_per_drone))
                allowed_users = attributes.get('users', None)
            else:
                disabled = config.get_config_value(
                        section, '%s_disabled' % hostname, default='')
                drone.enabled = not bool(disabled)
                drone.max_processes = config.get_config_value(
                        section, '%s_max_processes' % hostname, type=int,
                        default=scheduler_config.config.max_processes_per_drone)

                allowed_users = config.get_config_value(
                        section, '%s_users' % hostname, default=None)
            if allowed_users:
                drone.allowed_users = set(allowed_users.split())
            else:
                drone.allowed_users = None
            logging.info('Drone %s.max_processes: %s', hostname,
                         drone.max_processes)
            logging.info('Drone %s.enabled: %s', hostname, drone.enabled)
            logging.info('Drone %s.allowed_users: %s', hostname,
                         drone.allowed_users)

        self._reorder_drone_queue() # max_processes may have changed
        # Clear notification record about reaching max_processes limit.
        self._notify_record = {}


    def get_drones(self):
        return self._drones.itervalues()


    def cleanup_orphaned_containers(self):
        """Queue cleanup_orphaned_containers call at each drone.
        """
        for drone in self._drones.values():
            logging.info('Queue cleanup_orphaned_containers at %s',
                         drone.hostname)
            drone.queue_call('cleanup_orphaned_containers')


    def _get_drone_for_process(self, process):
        return self._drones[process.hostname]


    def _get_drone_for_pidfile_id(self, pidfile_id):
        pidfile_contents = self.get_pidfile_contents(pidfile_id)
        if pidfile_contents.process is None:
          raise DroneManagerError('Fail to get a drone due to empty pidfile')
        return self._get_drone_for_process(pidfile_contents.process)


    def get_drone_for_pidfile_id(self, pidfile_id):
        """Public API for luciferlib.

        @param pidfile_id: PidfileId instance.
        """
        return self._get_drone_for_pidfile_id(pidfile_id)


    def _drop_old_pidfiles(self):
        # use items() since the dict is modified in unregister_pidfile()
        for pidfile_id, info in self._registered_pidfile_info.items():
            if info.age > self._get_max_pidfile_refreshes():
                logging.warning('dropping leaked pidfile %s', pidfile_id)
                self.unregister_pidfile(pidfile_id)
            else:
                info.age += 1


    def _reset(self):
        self._process_set = set()
        self._all_processes = {}
        self._pidfiles = {}
        self._pidfiles_second_read = {}
        self._drone_queue = []


    def _parse_pidfile(self, drone, raw_contents):
        """Parse raw pidfile contents.

        @param drone: The drone on which this pidfile was found.
        @param raw_contents: The raw contents of a pidfile, eg:
            "pid\nexit_staus\nnum_tests_failed\n".
        """
        contents = PidfileContents()
        if not raw_contents:
            return contents
        lines = raw_contents.splitlines()
        if len(lines) > 3:
            return InvalidPidfile('Corrupt pid file (%d lines):\n%s' %
                                  (len(lines), lines))
        try:
            pid = int(lines[0])
            contents.process = Process(drone.hostname, pid)
            # if len(lines) == 2, assume we caught Autoserv between writing
            # exit_status and num_failed_tests, so just ignore it and wait for
            # the next cycle
            if len(lines) == 3:
                contents.exit_status = int(lines[1])
                contents.num_tests_failed = int(lines[2])
        except ValueError, exc:
            return InvalidPidfile('Corrupt pid file: ' + str(exc.args))

        return contents


    def _process_pidfiles(self, drone, pidfiles, store_in_dict):
        for pidfile_path, contents in pidfiles.iteritems():
            pidfile_id = PidfileId(pidfile_path)
            contents = self._parse_pidfile(drone, contents)
            store_in_dict[pidfile_id] = contents


    def _add_process(self, drone, process_info):
        process = Process(drone.hostname, int(process_info['pid']),
                          int(process_info['ppid']))
        self._process_set.add(process)


    def _add_autoserv_process(self, drone, process_info):
        assert process_info['comm'] == 'autoserv'
        # only root autoserv processes have pgid == pid
        if process_info['pgid'] != process_info['pid']:
            return
        self._add_process(drone, process_info)


    def _enqueue_drone(self, drone):
        heapq.heappush(self._drone_queue, _DroneHeapWrapper(drone))


    def _reorder_drone_queue(self):
        heapq.heapify(self._drone_queue)


    def reorder_drone_queue(self):
        """Reorder drone queue according to modified process counts.

        This public API is exposed for luciferlib to wrap.
        """
        self._reorder_drone_queue()


    def _compute_active_processes(self, drone):
        drone.active_processes = 0
        for pidfile_id, contents in self._pidfiles.iteritems():
            is_running = contents.exit_status is None
            on_this_drone = (contents.process
                             and contents.process.hostname == drone.hostname)
            if is_running and on_this_drone:
                info = self._registered_pidfile_info[pidfile_id]
                if info.num_processes is not None:
                    drone.active_processes += info.num_processes

        metrics.Gauge('chromeos/autotest/drone/active_processes').set(
                drone.active_processes,
                fields={'drone_hostname': drone.hostname})


    def _check_drone_process_limit(self, drone):
        """
        Notify if the number of processes on |drone| is approaching limit.

        @param drone: A Drone object.
        """
        try:
            percent = float(drone.active_processes) / drone.max_processes
        except ZeroDivisionError:
            percent = 100
        metrics.Float('chromeos/autotest/drone/active_process_percentage'
                      ).set(percent, fields={'drone_hostname': drone.hostname})

    def trigger_refresh(self):
        """Triggers a drone manager refresh.

        @raises DroneManagerError: If a drone has un-executed calls.
            Since they will get clobbered when we queue refresh calls.
        """
        self._reset()
        self._drop_old_pidfiles()
        pidfile_paths = [pidfile_id.path
                         for pidfile_id in self._registered_pidfile_info]
        drones = list(self.get_drones())
        for drone in drones:
            calls = drone.get_calls()
            if calls:
                raise DroneManagerError('Drone %s has un-executed calls: %s '
                                        'which might get corrupted through '
                                        'this invocation' %
                                        (drone, [str(call) for call in calls]))
            drone.queue_call('refresh', pidfile_paths)
        logging.info("Invoking drone refresh.")
        with metrics.SecondsTimer(
                'chromeos/autotest/drone_manager/trigger_refresh_duration'):
            self._refresh_task_queue.execute(drones, wait=False)


    def sync_refresh(self):
        """Complete the drone refresh started by trigger_refresh.

        Waits for all drone threads then refreshes internal datastructures
        with drone process information.
        """

        # This gives us a dictionary like what follows:
        # {drone: [{'pidfiles': (raw contents of pidfile paths),
        #           'autoserv_processes': (autoserv process info from ps),
        #           'all_processes': (all process info from ps),
        #           'parse_processes': (parse process infor from ps),
        #           'pidfile_second_read': (pidfile contents, again),}]
        #   drone2: ...}
        # The values of each drone are only a list because this adheres to the
        # drone utility interface (each call is executed and its results are
        # places in a list, but since we never couple the refresh calls with
        # any other call, this list will always contain a single dict).
        with metrics.SecondsTimer(
                'chromeos/autotest/drone_manager/sync_refresh_duration'):
            all_results = self._refresh_task_queue.get_results()
        logging.info("Drones refreshed.")

        # The loop below goes through and parses pidfile contents. Pidfiles
        # are used to track autoserv execution, and will always contain < 3
        # lines of the following: pid, exit code, number of tests. Each pidfile
        # is identified by a PidfileId object, which contains a unique pidfile
        # path (unique because it contains the job id) making it hashable.
        # All pidfiles are stored in the drone managers _pidfiles dict as:
        #   {pidfile_id: pidfile_contents(Process(drone, pid),
        #                                 exit_code, num_tests_failed)}
        # In handle agents, each agent knows its pidfile_id, and uses this
        # to retrieve the refreshed contents of its pidfile via the
        # PidfileRunMonitor (through its tick) before making decisions. If
        # the agent notices that its process has exited, it unregisters the
        # pidfile from the drone_managers._registered_pidfile_info dict
        # through its epilog.
        for drone, results_list in all_results.iteritems():
            results = results_list[0]
            drone_hostname = drone.hostname.replace('.', '_')

            for process_info in results['all_processes']:
                if process_info['comm'] == 'autoserv':
                    self._add_autoserv_process(drone, process_info)
                drone_pid = drone.hostname, int(process_info['pid'])
                self._all_processes[drone_pid] = process_info

            for process_info in results['parse_processes']:
                self._add_process(drone, process_info)

            self._process_pidfiles(drone, results['pidfiles'], self._pidfiles)
            self._process_pidfiles(drone, results['pidfiles_second_read'],
                                   self._pidfiles_second_read)

            self._compute_active_processes(drone)
            if drone.enabled:
                self._enqueue_drone(drone)
                self._check_drone_process_limit(drone)


    def refresh(self):
        """Refresh all drones."""
        with metrics.SecondsTimer(
                'chromeos/autotest/drone_manager/refresh_duration'):
            self.trigger_refresh()
            self.sync_refresh()


    @metrics.SecondsTimerDecorator(
        'chromeos/autotest/drone_manager/execute_actions_duration')
    def execute_actions(self):
        """
        Called at the end of a scheduler cycle to execute all queued actions
        on drones.
        """
        # Invoke calls queued on all drones since the last call to execute
        # and wait for them to return.
        if _THREADED_DRONE_MANAGER:
            thread_lib.ThreadedTaskQueue(
                    name='%s.execute_queue' % self._STATS_KEY).execute(
                            self._drones.values())
        else:
            drone_task_queue.DroneTaskQueue().execute(self._drones.values())

        try:
            self._results_drone.execute_queued_calls()
        except error.AutoservError:
            m = 'chromeos/autotest/errors/results_repository_failed'
            metrics.Counter(m).increment(
                fields={'drone_hostname': self._results_drone.hostname})
            self._results_drone.clear_call_queue()


    def get_orphaned_autoserv_processes(self):
        """
        Returns a set of Process objects for orphaned processes only.
        """
        return set(process for process in self._process_set
                   if process.ppid == 1)


    def kill_process(self, process):
        """
        Kill the given process.
        """
        logging.info('killing %s', process)
        drone = self._get_drone_for_process(process)
        drone.queue_kill_process(process)


    def _ensure_directory_exists(self, path):
        if not os.path.exists(path):
            os.makedirs(path)


    def total_running_processes(self):
        return sum(drone.active_processes for drone in self.get_drones())


    def max_runnable_processes(self, username, drone_hostnames_allowed):
        """
        Return the maximum number of processes that can be run (in a single
        execution) given the current load on drones.
        @param username: login of user to run a process.  may be None.
        @param drone_hostnames_allowed: list of drones that can be used. May be
                                        None
        """
        usable_drone_wrappers = [wrapper for wrapper in self._drone_queue
                                 if wrapper.drone.usable_by(username) and
                                 (drone_hostnames_allowed is None or
                                          wrapper.drone.hostname in
                                                  drone_hostnames_allowed)]
        if not usable_drone_wrappers:
            # all drones disabled or inaccessible
            return 0
        runnable_processes = [
                wrapper.drone.max_processes - wrapper.drone.active_processes
                for wrapper in usable_drone_wrappers]
        return max([0] + runnable_processes)


    def _least_loaded_drone(self, drones):
        return min(drones, key=lambda d: d.used_capacity())


    def pick_drone_to_use(self, num_processes=1):
        """Return a drone to use.

        Various options can be passed to optimize drone selection.

        num_processes is the number of processes the drone is intended
        to run.

        This public API is exposed for luciferlib to wrap.

        Returns a drone instance (see drones.py).
        """
        return self._choose_drone_for_execution(
                num_processes=num_processes,
                username=None,  # Always allow all drones
                drone_hostnames_allowed=None,  # Always allow all drones
        )


    def _choose_drone_for_execution(self, num_processes, username,
                                    drone_hostnames_allowed):
        """Choose a drone to execute command.

        @param num_processes: Number of processes needed for execution.
        @param username: Name of the user to execute the command.
        @param drone_hostnames_allowed: A list of names of drone allowed.

        @return: A drone object to be used for execution.
        """
        # cycle through drones is order of increasing used capacity until
        # we find one that can handle these processes
        checked_drones = []
        usable_drones = []
        drone_to_use = None
        while self._drone_queue:
            drone = heapq.heappop(self._drone_queue).drone
            checked_drones.append(drone)
            logging.info('Checking drone %s', drone.hostname)
            if not drone.usable_by(username):
                continue

            drone_allowed = (drone_hostnames_allowed is None
                             or drone.hostname in drone_hostnames_allowed)
            if not drone_allowed:
                logging.debug('Drone %s not allowed: ', drone.hostname)
                continue

            usable_drones.append(drone)

            if drone.active_processes + num_processes <= drone.max_processes:
                drone_to_use = drone
                break
            logging.info('Drone %s has %d active + %s requested > %s max',
                         drone.hostname, drone.active_processes, num_processes,
                         drone.max_processes)

        if not drone_to_use and usable_drones:
            # Drones are all over loaded, pick the one with least load.
            drone_summary = ','.join('%s %s/%s' % (drone.hostname,
                                                   drone.active_processes,
                                                   drone.max_processes)
                                     for drone in usable_drones)
            logging.error('No drone has capacity to handle %d processes (%s) '
                          'for user %s', num_processes, drone_summary, username)
            drone_to_use = self._least_loaded_drone(usable_drones)

        # refill _drone_queue
        for drone in checked_drones:
            self._enqueue_drone(drone)

        return drone_to_use


    def _substitute_working_directory_into_command(self, command,
                                                   working_directory):
        for i, item in enumerate(command):
            if item is WORKING_DIRECTORY:
                command[i] = working_directory


    def execute_command(self, command, working_directory, pidfile_name,
                        num_processes, log_file=None, paired_with_pidfile=None,
                        username=None, drone_hostnames_allowed=None):
        """
        Execute the given command, taken as an argv list.

        @param command: command to execute as a list.  if any item is
                WORKING_DIRECTORY, the absolute path to the working directory
                will be substituted for it.
        @param working_directory: directory in which the pidfile will be written
        @param pidfile_name: name of the pidfile this process will write
        @param num_processes: number of processes to account for from this
                execution
        @param log_file (optional): path (in the results repository) to hold
                command output.
        @param paired_with_pidfile (optional): a PidfileId for an
                already-executed process; the new process will execute on the
                same drone as the previous process.
        @param username (optional): login of the user responsible for this
                process.
        @param drone_hostnames_allowed (optional): hostnames of the drones that
                                                   this command is allowed to
                                                   execute on
        """
        abs_working_directory = self.absolute_path(working_directory)
        if not log_file:
            log_file = self.get_temporary_path('execute')
        log_file = self.absolute_path(log_file)

        self._substitute_working_directory_into_command(command,
                                                        abs_working_directory)

        if paired_with_pidfile:
            drone = self._get_drone_for_pidfile_id(paired_with_pidfile)
        else:
            drone = self._choose_drone_for_execution(
                    num_processes, username, drone_hostnames_allowed)

        if not drone:
            raise DroneManagerError('command failed; no drones available: %s'
                                    % command)

        logging.info("command = %s", command)
        logging.info('log file = %s:%s', drone.hostname, log_file)
        self._write_attached_files(working_directory, drone)
        drone.queue_call('execute_command', command, abs_working_directory,
                         log_file, pidfile_name)
        drone.active_processes += num_processes
        self._reorder_drone_queue()

        pidfile_path = os.path.join(abs_working_directory, pidfile_name)
        pidfile_id = PidfileId(pidfile_path)
        self.register_pidfile(pidfile_id)
        self._registered_pidfile_info[pidfile_id].num_processes = num_processes
        return pidfile_id


    def get_pidfile_id_from(self, execution_tag, pidfile_name):
        path = os.path.join(self.absolute_path(execution_tag), pidfile_name)
        return PidfileId(path)


    def register_pidfile(self, pidfile_id):
        """
        Indicate that the DroneManager should look for the given pidfile when
        refreshing.
        """
        if pidfile_id not in self._registered_pidfile_info:
            logging.info('monitoring pidfile %s', pidfile_id)
            self._registered_pidfile_info[pidfile_id] = _PidfileInfo()
        self._reset_pidfile_age(pidfile_id)


    def _reset_pidfile_age(self, pidfile_id):
        if pidfile_id in self._registered_pidfile_info:
            self._registered_pidfile_info[pidfile_id].age = 0


    def unregister_pidfile(self, pidfile_id):
        if pidfile_id in self._registered_pidfile_info:
            logging.info('forgetting pidfile %s', pidfile_id)
            del self._registered_pidfile_info[pidfile_id]


    def declare_process_count(self, pidfile_id, num_processes):
        self._registered_pidfile_info[pidfile_id].num_processes = num_processes


    def get_pidfile_contents(self, pidfile_id, use_second_read=False):
        """
        Retrieve a PidfileContents object for the given pidfile_id.  If
        use_second_read is True, use results that were read after the processes
        were checked, instead of before.
        """
        self._reset_pidfile_age(pidfile_id)
        if use_second_read:
            pidfile_map = self._pidfiles_second_read
        else:
            pidfile_map = self._pidfiles
        return pidfile_map.get(pidfile_id, PidfileContents())


    def is_process_running(self, process):
        """
        Check if the given process is in the running process list.
        """
        if process in self._process_set:
            return True

        drone_pid = process.hostname, process.pid
        if drone_pid in self._all_processes:
            logging.error('Process %s found, but not an autoserv process. '
                    'Is %s', process, self._all_processes[drone_pid])
            return True

        return False


    def get_temporary_path(self, base_name):
        """
        Get a new temporary path guaranteed to be unique across all drones
        for this scheduler execution.
        """
        self._temporary_path_counter += 1
        return os.path.join(drone_utility._TEMPORARY_DIRECTORY,
                            '%s.%s' % (base_name, self._temporary_path_counter))


    def absolute_path(self, path, on_results_repository=False):
        if on_results_repository:
            base_dir = self._results_dir
        else:
            base_dir = os.path.join(drones.AUTOTEST_INSTALL_DIR,
                                    _DRONE_RESULTS_DIR_SUFFIX)
        return os.path.join(base_dir, path)


    def _copy_results_helper(self, process, source_path, destination_path,
                             to_results_repository=False):
        logging.debug('_copy_results_helper. process: %s, source_path: %s, '
                      'destination_path: %s, to_results_repository: %s',
                      process, source_path, destination_path,
                      to_results_repository)
        full_source = self.absolute_path(source_path)
        full_destination = self.absolute_path(
                destination_path, on_results_repository=to_results_repository)
        source_drone = self._get_drone_for_process(process)
        if to_results_repository:
            source_drone.send_file_to(self._results_drone, full_source,
                                      full_destination, can_fail=True)
        else:
            source_drone.queue_call('copy_file_or_directory', full_source,
                                    full_destination)


    def copy_to_results_repository(self, process, source_path,
                                   destination_path=None):
        """
        Copy results from the given process at source_path to destination_path
        in the results repository.

        This will only copy the results back for Special Agent Tasks (Cleanup,
        Verify, Repair) that reside in the hosts/ subdirectory of results if
        the copy_task_results_back flag has been set to True inside
        global_config.ini

        It will also only copy .parse.log files back to the scheduler if the
        copy_parse_log_back flag in global_config.ini has been set to True.
        """
        if not ENABLE_ARCHIVING:
            return
        copy_task_results_back = global_config.global_config.get_config_value(
                scheduler_config.CONFIG_SECTION, 'copy_task_results_back',
                type=bool)
        copy_parse_log_back = global_config.global_config.get_config_value(
                scheduler_config.CONFIG_SECTION, 'copy_parse_log_back',
                type=bool)
        special_task = source_path.startswith(HOSTS_JOB_SUBDIR)
        parse_log = source_path.endswith(PARSE_LOG)
        if (copy_task_results_back or not special_task) and (
                copy_parse_log_back or not parse_log):
            if destination_path is None:
                destination_path = source_path
            self._copy_results_helper(process, source_path, destination_path,
                                      to_results_repository=True)

    def _copy_to_results_repository(self, process, source_path,
                                   destination_path=None):
        """
        Copy results from the given process at source_path to destination_path
        in the results repository, without special task handling.
        """
        if destination_path is None:
            destination_path = source_path
        self._copy_results_helper(process, source_path, destination_path,
                                  to_results_repository=True)


    def copy_results_on_drone(self, process, source_path, destination_path):
        """
        Copy a results directory from one place to another on the drone.
        """
        self._copy_results_helper(process, source_path, destination_path)


    def _write_attached_files(self, results_dir, drone):
        attached_files = self._attached_files.pop(results_dir, {})
        for file_path, contents in attached_files.iteritems():
            drone.queue_call('write_to_file', self.absolute_path(file_path),
                             contents)


    def attach_file_to_execution(self, results_dir, file_contents,
                                 file_path=None):
        """
        When the process for the results directory is executed, the given file
        contents will be placed in a file on the drone.  Returns the path at
        which the file will be placed.
        """
        if not file_path:
            file_path = self.get_temporary_path('attach')
        files_for_execution = self._attached_files.setdefault(results_dir, {})
        assert file_path not in files_for_execution
        files_for_execution[file_path] = file_contents
        return file_path


    def write_lines_to_file(self, file_path, lines, paired_with_process=None):
        """
        Write the given lines (as a list of strings) to a file.  If
        paired_with_process is given, the file will be written on the drone
        running the given Process.  Otherwise, the file will be written to the
        results repository.
        """
        file_contents = '\n'.join(lines) + '\n'
        if paired_with_process:
            drone = self._get_drone_for_process(paired_with_process)
            on_results_repository = False
        else:
            drone = self._results_drone
            on_results_repository = True
        full_path = self.absolute_path(
                file_path, on_results_repository=on_results_repository)
        drone.queue_call('write_to_file', full_path, file_contents)


_the_instance = None

def instance():
    if _the_instance is None:
        _set_instance(DroneManager())
    return _the_instance


def _set_instance(instance): # usable for testing
    global _the_instance
    _the_instance = instance