scheduler/query_managers.py

#pylint: disable-msg=C0111

# Copyright (c) 2014 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""Scheduler library classes.
"""

import collections
import logging

import common

from autotest_lib.client.common_lib.cros.graphite import stats
from autotest_lib.frontend import setup_django_environment
from autotest_lib.frontend.afe import models
from autotest_lib.server.cros.dynamic_suite import constants
from autotest_lib.scheduler import scheduler_models
from autotest_lib.scheduler import scheduler_lib


_job_timer = stats.Timer('scheduler.job_query_manager')
class AFEJobQueryManager(object):
    """Query manager for AFE Jobs."""

    # A subquery to only get inactive hostless jobs.
    hostless_query = 'host_id IS NULL AND meta_host IS NULL'


    @_job_timer.decorate
    def get_pending_queue_entries(self, only_hostless=False):
        """
        Fetch a list of new host queue entries.

        The ordering of this list is important, as every new agent
        we schedule can potentially contribute to the process count
        on the drone, which has a static limit. The sort order
        prioritizes jobs as follows:
        1. High priority jobs: Based on the afe_job's priority
        2. With hosts and metahosts: This will only happen if we don't
            activate the hqe after assigning a host to it in
            schedule_new_jobs.
        3. With hosts but without metahosts: When tests are scheduled
            through the frontend the owner of the job would have chosen
            a host for it.
        4. Without hosts but with metahosts: This is the common case of
            a new test that needs a DUT. We assign a host and set it to
            active so it shouldn't show up in case 2 on the next tick.
        5. Without hosts and without metahosts: Hostless suite jobs, that
            will result in new jobs that fall under category 4.

        A note about the ordering of cases 3 and 4:
        Prioritizing one case above the other leads to earlier acquisition
        of the following resources: 1. process slots on the drone 2. machines.
        - When a user schedules a job through the afe they choose a specific
          host for it. Jobs with metahost can utilize any host that satisfies
          the metahost criterion. This means that if we had scheduled 4 before
          3 there is a good chance that a job which could've used another host,
          will now use the host assigned to a metahost-less job. Given the
          availability of machines in pool:suites, this almost guarantees
          starvation for jobs scheduled through the frontend.
        - Scheduling 4 before 3 also has its pros however, since a suite
          has the concept of a time out, whereas users can wait. If we hit the
          process count on the drone a suite can timeout waiting on the test,
          but a user job generally has a much longer timeout, and relatively
          harmless consequences.
        The current ordering was chosed because it is more likely that we will
        run out of machines in pool:suites than processes on the drone.

        @returns A list of HQEs ordered according to sort_order.
        """
        sort_order = ('afe_jobs.priority DESC, '
                      'ISNULL(host_id), '
                      'ISNULL(meta_host), '
                      'parent_job_id, '
                      'job_id')
        # Don't execute jobs that should be executed by a shard in the global
        # scheduler.
        # This won't prevent the shard scheduler to run this, as the shard db
        # doesn't have an an entry in afe_shards_labels.
        query=('NOT complete AND NOT active AND status="Queued"'
               'AND NOT aborted AND afe_shards_labels.id IS NULL')

        # TODO(jakobjuelich, beeps): Optimize this query. Details:
        # Compressed output of EXPLAIN <query>:
        # +------------------------+--------+-------------------------+-------+
        # | table                  | type   | key                     | rows  |
        # +------------------------+--------+-------------------------+-------+
        # | afe_host_queue_entries | ref    | host_queue_entry_status | 30536 |
        # | afe_shards_labels      | ref    | shard_label_id_fk       |     1 |
        # | afe_jobs               | eq_ref | PRIMARY                 |     1 |
        # +------------------------+--------+-------------------------+-------+
        # This shows the first part of the query fetches a lot of objects, that
        # are then filtered. The joins are comparably fast: There's usually just
        # one or none shard mapping that can be answered fully using an index
        # (shard_label_id_fk), similar thing applies to the job.
        #
        # This works for now, but once O(#Jobs in shard) << O(#Jobs in Queued),
        # it might be more efficient to filter on the meta_host first, instead
        # of the status.
        if only_hostless:
            query = '%s AND (%s)' % (query, self.hostless_query)
        return list(scheduler_models.HostQueueEntry.fetch(
            joins=('INNER JOIN afe_jobs ON (job_id=afe_jobs.id) '
                   'LEFT JOIN afe_shards_labels ON ('
                   'meta_host=afe_shards_labels.label_id)'),
            where=query, order_by=sort_order))


    @_job_timer.decorate
    def get_prioritized_special_tasks(self, only_tasks_with_leased_hosts=False):
        """
        Returns all queued SpecialTasks prioritized for repair first, then
        cleanup, then verify.

        @param only_tasks_with_leased_hosts: If true, this method only returns
            tasks with leased hosts.

        @return: list of afe.models.SpecialTasks sorted according to priority.
        """
        queued_tasks = models.SpecialTask.objects.filter(is_active=False,
                                                         is_complete=False,
                                                         host__locked=False)
        # exclude hosts with active queue entries unless the SpecialTask is for
        # that queue entry
        queued_tasks = models.SpecialTask.objects.add_join(
                queued_tasks, 'afe_host_queue_entries', 'host_id',
                join_condition='afe_host_queue_entries.active',
                join_from_key='host_id', force_left_join=True)
        queued_tasks = queued_tasks.extra(
                where=['(afe_host_queue_entries.id IS NULL OR '
                       'afe_host_queue_entries.id = '
                               'afe_special_tasks.queue_entry_id)'])
        if only_tasks_with_leased_hosts:
            queued_tasks = queued_tasks.filter(host__leased=True)

        # reorder tasks by priority
        task_priority_order = [models.SpecialTask.Task.REPAIR,
                               models.SpecialTask.Task.CLEANUP,
                               models.SpecialTask.Task.VERIFY,
                               models.SpecialTask.Task.RESET,
                               models.SpecialTask.Task.PROVISION]
        def task_priority_key(task):
            return task_priority_order.index(task.task)
        return sorted(queued_tasks, key=task_priority_key)


    @classmethod
    def get_overlapping_jobs(cls):
        """A helper method to get all active jobs using the same host.

        @return: A list of dictionaries with the hqe id, job_id and host_id
            of the currently overlapping jobs.
        """
        # Filter all active hqes and stand alone special tasks to make sure
        # a host isn't being used by two jobs at the same time. An incomplete
        # stand alone special task can share a host with an active hqe, an
        # example of this is the cleanup scheduled in gathering.
        hqe_hosts = list(models.HostQueueEntry.objects.filter(
                active=1, complete=0, host_id__isnull=False).values_list(
                'host_id', flat=True))
        special_task_hosts = list(models.SpecialTask.objects.filter(
                is_active=1, is_complete=0, host_id__isnull=False,
                queue_entry_id__isnull=True).values_list('host_id', flat=True))
        host_counts = collections.Counter(
                hqe_hosts + special_task_hosts).most_common()
        multiple_hosts = [count[0] for count in host_counts if count[1] > 1]
        return list(models.HostQueueEntry.objects.filter(
                host_id__in=multiple_hosts, active=True).values(
                        'id', 'job_id', 'host_id'))


    @_job_timer.decorate
    def get_suite_host_assignment(self):
        """A helper method to get how many hosts each suite is holding.

        @return: Two dictionaries (suite_host_num, hosts_to_suites)
                 suite_host_num maps suite job id to number of hosts
                 holding by its child jobs.
                 hosts_to_suites contains current hosts held by
                 any suites, and maps the host id to its parent_job_id.
        """
        query = models.HostQueueEntry.objects.filter(
                host_id__isnull=False, complete=0, active=1,
                job__parent_job_id__isnull=False)
        suite_host_num = {}
        hosts_to_suites = {}
        for hqe in query:
            host_id = hqe.host_id
            parent_job_id = hqe.job.parent_job_id
            count = suite_host_num.get(parent_job_id, 0)
            suite_host_num[parent_job_id] = count + 1
            hosts_to_suites[host_id] = parent_job_id
        return suite_host_num, hosts_to_suites


    @_job_timer.decorate
    def get_min_duts_of_suites(self, suite_job_ids):
        """Load suite_min_duts job keyval for a set of suites.

        @param suite_job_ids: A set of suite job ids.

        @return: A dictionary where the key is a suite job id,
                 the value is the value of 'suite_min_duts'.
        """
        query = models.JobKeyval.objects.filter(
                job_id__in=suite_job_ids,
                key=constants.SUITE_MIN_DUTS_KEY, value__isnull=False)
        return dict((keyval.job_id, int(keyval.value)) for keyval in query)


_host_timer = stats.Timer('scheduler.host_query_manager')
class AFEHostQueryManager(object):
    """Query manager for AFE Hosts."""

    def __init__(self):
        """Create an AFEHostQueryManager.

        @param db: A connection to the database with the afe_hosts table.
        """
        self._db = scheduler_lib.ConnectionManager().get_connection()


    def _process_many2many_dict(self, rows, flip=False):
        result = {}
        for row in rows:
            left_id, right_id = int(row[0]), int(row[1])
            if flip:
                left_id, right_id = right_id, left_id
            result.setdefault(left_id, set()).add(right_id)
        return result


    def _get_sql_id_list(self, id_list):
        return ','.join(str(item_id) for item_id in id_list)


    def _get_many2many_dict(self, query, id_list, flip=False):
        if not id_list:
            return {}
        query %= self._get_sql_id_list(id_list)
        rows = self._db.execute(query)
        return self._process_many2many_dict(rows, flip)


    @_host_timer.decorate
    def _get_ready_hosts(self):
        # We don't lose anything by re-doing these checks
        # even though we release hosts on the same conditions.
        # In the future we might have multiple clients that
        # release_hosts and/or lock them independent of the
        # scheduler tick.
        hosts = scheduler_models.Host.fetch(
            where="NOT afe_hosts.leased "
                  "AND NOT afe_hosts.locked "
                  "AND (afe_hosts.status IS NULL "
                      "OR afe_hosts.status = 'Ready')")
        return dict((host.id, host) for host in hosts)


    @_host_timer.decorate
    def _get_job_acl_groups(self, job_ids):
        query = """
        SELECT afe_jobs.id, afe_acl_groups_users.aclgroup_id
        FROM afe_jobs
        INNER JOIN afe_users ON afe_users.login = afe_jobs.owner
        INNER JOIN afe_acl_groups_users ON
                afe_acl_groups_users.user_id = afe_users.id
        WHERE afe_jobs.id IN (%s)
        """
        return self._get_many2many_dict(query, job_ids)


    @_host_timer.decorate
    def _get_job_ineligible_hosts(self, job_ids):
        query = """
        SELECT job_id, host_id
        FROM afe_ineligible_host_queues
        WHERE job_id IN (%s)
        """
        return self._get_many2many_dict(query, job_ids)


    @_host_timer.decorate
    def _get_job_dependencies(self, job_ids):
        query = """
        SELECT job_id, label_id
        FROM afe_jobs_dependency_labels
        WHERE job_id IN (%s)
        """
        return self._get_many2many_dict(query, job_ids)

    @_host_timer.decorate
    def _get_host_acls(self, host_ids):
        query = """
        SELECT host_id, aclgroup_id
        FROM afe_acl_groups_hosts
        WHERE host_id IN (%s)
        """
        return self._get_many2many_dict(query, host_ids)


    @_host_timer.decorate
    def _get_label_hosts(self, host_ids):
        if not host_ids:
            return {}, {}
        query = """
        SELECT label_id, host_id
        FROM afe_hosts_labels
        WHERE host_id IN (%s)
        """ % self._get_sql_id_list(host_ids)
        rows = self._db.execute(query)
        labels_to_hosts = self._process_many2many_dict(rows)
        hosts_to_labels = self._process_many2many_dict(rows, flip=True)
        return labels_to_hosts, hosts_to_labels


    @classmethod
    def find_unused_healty_hosts(cls):
        """Get hosts that are currently unused and in the READY state.

        @return: A list of host objects, one for each unused healthy host.
        """
        # Avoid any host with a currently active queue entry against it.
        hqe_join = ('LEFT JOIN afe_host_queue_entries AS active_hqe '
                    'ON (afe_hosts.id = active_hqe.host_id AND '
                    'active_hqe.active)')

        # Avoid any host with a new special task against it. There are 2 cases
        # when an inactive but incomplete special task will not use the host
        # this tick: 1. When the host is locked 2. When an active hqe already
        # has special tasks for the same host. In both these cases this host
        # will not be in the ready hosts list anyway. In all other cases,
        # an incomplete special task will grab the host before a new job does
        # by assigning an agent to it.
        special_task_join = ('LEFT JOIN afe_special_tasks as new_tasks '
                             'ON (afe_hosts.id = new_tasks.host_id AND '
                             'new_tasks.is_complete=0)')

        return scheduler_models.Host.fetch(
            joins='%s %s' % (hqe_join, special_task_join),
            where="active_hqe.host_id IS NULL AND new_tasks.host_id IS NULL "
                  "AND afe_hosts.leased "
                  "AND NOT afe_hosts.locked "
                  "AND (afe_hosts.status IS NULL "
                          "OR afe_hosts.status = 'Ready')")


    @_host_timer.decorate
    def set_leased(self, leased_value, **kwargs):
        """Modify the leased bit on the hosts with ids in host_ids.

        @param leased_value: The True/False value of the leased column for
            the hosts with ids in host_ids.
        @param kwargs: The args to use in finding matching hosts.
        """
        logging.info('Setting leased = %s for the hosts that match %s',
                     leased_value, kwargs)
        models.Host.objects.filter(**kwargs).update(leased=leased_value)


    @_host_timer.decorate
    def _get_labels(self, job_dependencies):
        """
        Calculate a dict mapping label id to label object so that we don't
        frequently round trip to the database every time we need a label.

        @param job_dependencies: A dict mapping an integer job id to a list of
            integer label id's.  ie. {job_id: [label_id]}
        @return: A dict mapping an integer label id to a scheduler model label
            object.  ie. {label_id: label_object}

        """
        id_to_label = dict()
        # Pull all the labels on hosts we might look at
        host_labels = scheduler_models.Label.fetch(
                where="id IN (SELECT label_id FROM afe_hosts_labels)")
        id_to_label.update([(label.id, label) for label in host_labels])
        # and pull all the labels on jobs we might look at.
        job_label_set = set()
        for job_deps in job_dependencies.values():
            job_label_set.update(job_deps)
        # On the rare/impossible chance that no jobs have any labels, we
        # can skip this.
        if job_label_set:
            job_string_label_list = ','.join([str(x) for x in job_label_set])
            job_labels = scheduler_models.Label.fetch(
                    where="id IN (%s)" % job_string_label_list)
            id_to_label.update([(label.id, label) for label in job_labels])
        return id_to_label


    @_host_timer.decorate
    def refresh(self, pending_queue_entries):
        """Update the query manager.

        Cache information about a list of queue entries and eligible hosts
        from the database so clients can avoid expensive round trips during
        host acquisition.

        @param pending_queue_entries: A list of queue entries about which we
            need information.
        """
        self._hosts_available = self._get_ready_hosts()
        relevant_jobs = [queue_entry.job_id
                         for queue_entry in pending_queue_entries]
        self._job_acls = self._get_job_acl_groups(relevant_jobs)
        self._ineligible_hosts = (self._get_job_ineligible_hosts(relevant_jobs))
        self._job_dependencies = (self._get_job_dependencies(relevant_jobs))
        host_ids = self._hosts_available.keys()
        self._host_acls = self._get_host_acls(host_ids)
        self._label_hosts, self._host_labels = (
                self._get_label_hosts(host_ids))
        self._labels = self._get_labels(self._job_dependencies)