321 lines
13 KiB
Python
321 lines
13 KiB
Python
# Copyright (c) 2011 OpenStack Foundation
|
|
# All Rights Reserved.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
|
# not use this file except in compliance with the License. You may obtain
|
|
# a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
|
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
|
# License for the specific language governing permissions and limitations
|
|
# under the License.
|
|
|
|
'''
|
|
Leverages nova/scheduler/filter_scheduler.py
|
|
'''
|
|
|
|
"""The FilterScheduler is for creating servers.
|
|
|
|
You can customize this scheduler by specifying your own node Filters and
|
|
Weighing Functions.
|
|
"""
|
|
import itertools
|
|
|
|
from oslo_config import cfg
|
|
from oslo_log import log as logging
|
|
|
|
from mogan.common import exception
|
|
from mogan.common.i18n import _
|
|
from mogan.common import utils
|
|
from mogan import objects
|
|
from mogan.scheduler import client
|
|
from mogan.scheduler import driver
|
|
from mogan.scheduler import utils as sched_utils
|
|
|
|
CONF = cfg.CONF
|
|
LOG = logging.getLogger(__name__)
|
|
|
|
|
|
class FilterScheduler(driver.Scheduler):
|
|
"""Scheduler that can be used for filtering and weighing."""
|
|
def __init__(self, *args, **kwargs):
|
|
super(FilterScheduler, self).__init__(*args, **kwargs)
|
|
self.max_attempts = self._max_attempts()
|
|
self.reportclient = client.SchedulerClient().reportclient
|
|
|
|
def _max_attempts(self):
|
|
max_attempts = CONF.scheduler.scheduler_max_attempts
|
|
if max_attempts < 1:
|
|
raise exception.InvalidParameterValue(
|
|
err=_("Invalid value for 'scheduler_max_attempts', "
|
|
"must be >=1"))
|
|
return max_attempts
|
|
|
|
def _log_server_error(self, server_id, retry):
|
|
"""Log requests with exceptions from previous server operations."""
|
|
exc = retry.pop('exc', None) # string-ified exception from server
|
|
if not exc:
|
|
return # no exception info from a previous attempt, skip
|
|
|
|
nodes = retry.get('nodes', None)
|
|
if not nodes:
|
|
return # no previously attempted nodes, skip
|
|
|
|
last_node = nodes[-1]
|
|
LOG.error("Error scheduling %(server_id)s from last node: "
|
|
"%(last_node)s : %(exc)s",
|
|
{'server_id': server_id,
|
|
'last_node': last_node,
|
|
'exc': exc})
|
|
|
|
def _populate_retry(self, filter_properties, request_spec):
|
|
"""Populate filter properties with history of retries for request.
|
|
|
|
If maximum retries is exceeded, raise NoValidNode.
|
|
"""
|
|
max_attempts = self.max_attempts
|
|
retry = filter_properties.pop('retry', {})
|
|
|
|
if max_attempts == 1:
|
|
# re-scheduling is disabled.
|
|
return
|
|
|
|
server_id = request_spec.get('server_ids')[0]
|
|
self._log_server_error(server_id, retry)
|
|
|
|
if retry['num_attempts'] > max_attempts:
|
|
raise exception.NoValidNode(
|
|
_("Exceeded max scheduling attempts %(max_attempts)d "
|
|
"for server %(server_id)s") %
|
|
{'max_attempts': max_attempts,
|
|
'server_id': server_id})
|
|
|
|
@staticmethod
|
|
def _get_res_cls_filters(request_spec):
|
|
flavor_dict = request_spec['flavor']
|
|
resources = dict([(sched_utils.ensure_resource_class_name(res[0]),
|
|
int(res[1]))
|
|
for res in flavor_dict['resources'].items()])
|
|
return resources
|
|
|
|
@staticmethod
|
|
def _get_res_aggregates_filters(context, request_spec):
|
|
flavor_dict = request_spec['flavor']
|
|
resource_aggregates = flavor_dict.get('resource_aggregates', {})
|
|
resource_aggregates_items = resource_aggregates.items()
|
|
# Add availability_zone aggregate
|
|
if request_spec['availability_zone']:
|
|
resource_aggregates_items.append(
|
|
('availability_zone', request_spec['availability_zone']))
|
|
|
|
filters = []
|
|
for key, value in resource_aggregates_items:
|
|
aggregates = objects.AggregateList.get_by_metadata(
|
|
context, key, value)
|
|
if not aggregates:
|
|
# if no aggregates match with the key/value,
|
|
# fail the scheduling.
|
|
return None
|
|
filters.append(aggregates)
|
|
|
|
return filters
|
|
|
|
@staticmethod
|
|
def _get_server_group_obj(context, request_spec):
|
|
server_group = request_spec.get('scheduler_hints', {}).get('group')
|
|
if not server_group:
|
|
return
|
|
server_group = objects.ServerGroup.get_by_uuid(context, server_group)
|
|
return server_group
|
|
|
|
def _get_nodes_of_aggregates(self, aggregates):
|
|
if not aggregates:
|
|
return []
|
|
agg_uuids = [agg.uuid for agg in aggregates]
|
|
query_filters = {'member_of': 'in:' + ','.join(agg_uuids)}
|
|
rps = self.reportclient.get_filtered_resource_providers(query_filters)
|
|
return [rp['uuid'] for rp in rps]
|
|
|
|
def _get_filtered_affzs_nodes(self, context, server_group, filtered_nodes,
|
|
num_servers):
|
|
"""Get the filtered affinity zone and nodes.
|
|
|
|
If affinity specified in request, this method will return a tuple
|
|
including filtered affinity zone and filtered nodes. e.g.
|
|
[(zone1, [node-1, node-2, node-3])].
|
|
If anti-affinity specified, this will return a list of tuples of
|
|
affinity zone and nodes list. e.g.
|
|
[(zone1, node-1]), (zone2, node-2), (zone3, node-3)]
|
|
|
|
"""
|
|
|
|
def _log_and_raise_error(policy):
|
|
LOG.error("No enough nodes filtered, request %(num_svr)s "
|
|
"server(s) with server group %(group)s with %(policy)s "
|
|
"policy specified.",
|
|
{"num_svr": num_servers, "group": server_group.name,
|
|
"policy": policy})
|
|
msg = (_("No enough nodes filtered, request %(num_svr)s server(s) "
|
|
"with %(policy)s policy specified.") %
|
|
{"num_svr": num_servers, "policy": policy})
|
|
raise exception.NoValidNode(msg)
|
|
|
|
if 'affinity' in server_group.policies:
|
|
selected_affz = None
|
|
if server_group.members:
|
|
for member in server_group.members:
|
|
server = objects.Server.get(context, member)
|
|
if server.affinity_zone:
|
|
selected_affz = server.affinity_zone
|
|
break
|
|
if selected_affz:
|
|
aggs = objects.AggregateList.get_by_metadata(
|
|
context, 'affinity_zone', selected_affz)
|
|
affz_nodes = self._get_nodes_of_aggregates(aggs)
|
|
selected_nodes = list(set(filtered_nodes) & set(affz_nodes))
|
|
if len(selected_nodes) < num_servers:
|
|
_log_and_raise_error('affinity')
|
|
return selected_affz, selected_nodes[:num_servers]
|
|
|
|
all_aggs = objects.AggregateList.get_by_metadata_key(
|
|
context, 'affinity_zone')
|
|
all_aggs = sorted(all_aggs, key=lambda a: a.metadata.get(
|
|
'affinity_zone'))
|
|
grouped_aggs = itertools.groupby(
|
|
all_aggs, lambda a: a.metadata.get('affinity_zone'))
|
|
|
|
if 'affinity' in server_group.policies:
|
|
for affz, aggs in grouped_aggs:
|
|
affz_nodes = self._get_nodes_of_aggregates(aggs)
|
|
affz_nodes = list(set(filtered_nodes) & set(affz_nodes))
|
|
if len(affz_nodes) >= num_servers:
|
|
return affz, affz_nodes[:num_servers]
|
|
_log_and_raise_error('affinity')
|
|
|
|
elif 'anti-affinity' in server_group.policies:
|
|
affinity_zones = []
|
|
for member in server_group.members:
|
|
server = objects.Server.get(context, member)
|
|
affinity_zone = server.affinity_zone
|
|
affinity_zones.append(affinity_zone)
|
|
selected_affz_nodes = []
|
|
for affz, aggs in grouped_aggs:
|
|
if affz in affinity_zones:
|
|
continue
|
|
affz_nodes = self._get_nodes_of_aggregates(aggs)
|
|
affz_nodes = list(set(filtered_nodes) & set(affz_nodes))
|
|
if affz_nodes:
|
|
selected_affz_nodes.append((affz, affz_nodes[0]))
|
|
if len(selected_affz_nodes) >= num_servers:
|
|
return selected_affz_nodes
|
|
_log_and_raise_error('anti-affinity')
|
|
|
|
def _consume_per_server(self, context, request_spec, node, server_id,
|
|
affinity_zone=None):
|
|
server_obj = objects.Server.get(context, server_id)
|
|
if affinity_zone:
|
|
server_obj.affinity_zone = affinity_zone
|
|
server_obj.save(context)
|
|
alloc_data = self._get_res_cls_filters(request_spec)
|
|
self.reportclient.put_allocations(
|
|
node, server_obj.uuid, alloc_data,
|
|
server_obj.project_id, server_obj.user_id)
|
|
|
|
def _consume_nodes_with_server_group(self, context, request_spec,
|
|
filtered_affzs_nodes, server_group):
|
|
if 'affinity' in server_group.policies:
|
|
affinity_zone, dest_nodes = filtered_affzs_nodes
|
|
for server_id, node in zip(request_spec['server_ids'],
|
|
dest_nodes):
|
|
self._consume_per_server(
|
|
context, request_spec, node, server_id, affinity_zone)
|
|
return dest_nodes
|
|
elif 'anti-affinity' in server_group.policies:
|
|
dest_nodes = []
|
|
for server_id, affz_node in zip(request_spec['server_ids'],
|
|
filtered_affzs_nodes):
|
|
affinity_zone, node = affz_node
|
|
dest_nodes.append(node)
|
|
self._consume_per_server(
|
|
context, request_spec, node, server_id, affinity_zone)
|
|
return dest_nodes
|
|
|
|
def _get_filtered_nodes(self, context, request_spec):
|
|
resources_filter = self._get_res_cls_filters(request_spec)
|
|
aggs_filters = self._get_res_aggregates_filters(context, request_spec)
|
|
|
|
# None indicates no matching aggregates
|
|
if aggs_filters is None:
|
|
return []
|
|
|
|
if aggs_filters:
|
|
filtered_nodes = set()
|
|
for agg_filter in aggs_filters:
|
|
filtered_rps = set(self._get_nodes_of_aggregates(agg_filter))
|
|
if not filtered_rps:
|
|
# if got empty, just break here.
|
|
return []
|
|
if not filtered_nodes:
|
|
# initialize the filtered_nodes
|
|
filtered_nodes = filtered_rps
|
|
else:
|
|
filtered_nodes &= filtered_rps
|
|
if not filtered_nodes:
|
|
# if got empty, just break here.
|
|
return []
|
|
else:
|
|
query_filters = {'resources': resources_filter}
|
|
filtered_nodes = self.reportclient.\
|
|
get_filtered_resource_providers(query_filters)
|
|
return [node['uuid'] for node in filtered_nodes]
|
|
|
|
return list(filtered_nodes)
|
|
|
|
def schedule(self, context, request_spec, filter_properties=None):
|
|
|
|
# TODO(zhenguo): Scheduler API is inherently multi-threaded as every
|
|
# incoming RPC message will be dispatched in it's own green thread.
|
|
# So we add a syncronized here to make sure the shared node states
|
|
# consistent, but lock the whole schedule process is not a good choice,
|
|
# we need to improve this.
|
|
@utils.synchronized('schedule')
|
|
def _schedule(self, context, request_spec, filter_properties):
|
|
self._populate_retry(filter_properties, request_spec)
|
|
filtered_nodes = self._get_filtered_nodes(context, request_spec)
|
|
|
|
if not filtered_nodes:
|
|
LOG.warning('No filtered nodes found for server '
|
|
'with properties: %s',
|
|
request_spec.get('flavor'))
|
|
raise exception.NoValidNode(
|
|
_("No filtered nodes available"))
|
|
dest_nodes = self._choose_nodes(filtered_nodes, request_spec)
|
|
server_group = self._get_server_group_obj(context, request_spec)
|
|
if not server_group:
|
|
for server_id, node in zip(request_spec['server_ids'],
|
|
dest_nodes):
|
|
self._consume_per_server(context, request_spec, node,
|
|
server_id)
|
|
return dest_nodes
|
|
else:
|
|
filtered_affzs_nodes = self._get_filtered_affzs_nodes(
|
|
context, server_group, filtered_nodes,
|
|
request_spec['num_servers'])
|
|
return self._consume_nodes_with_server_group(
|
|
context, request_spec, filtered_affzs_nodes, server_group)
|
|
|
|
return _schedule(self, context, request_spec, filter_properties)
|
|
|
|
def _choose_nodes(self, filtered_nodes, request_spec):
|
|
num_servers = request_spec['num_servers']
|
|
if num_servers > len(filtered_nodes):
|
|
msg = 'Not enough nodes found for servers, request ' \
|
|
'servers: %s, but only available nodes: %s' \
|
|
% (str(num_servers), str(len(filtered_nodes)))
|
|
raise exception.NoValidNode(_("Choose Node: %s") % msg)
|
|
|
|
return filtered_nodes[:num_servers]
|