| #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 autotest_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 = autotest_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 = autotest_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) |
