321 lines
13 KiB

# 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
# 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/
"""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
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.
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})
def _get_res_cls_filters(request_spec):
flavor_dict = request_spec['flavor']
resources = dict([(sched_utils.ensure_resource_class_name(res[0]),
for res in flavor_dict['resources'].items()])
return resources
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']:
('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
return filters
def _get_server_group_obj(context, request_spec):
server_group = request_spec.get('scheduler_hints', {}).get('group')
if not server_group:
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,
"""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":,
"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
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:
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(
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]
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
selected_affz_nodes = []
for affz, aggs in grouped_aggs:
if affz in affinity_zones:
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
def _consume_per_server(self, context, request_spec, node, server_id,
server_obj = objects.Server.get(context, server_id)
if affinity_zone:
server_obj.affinity_zone = affinity_zone
alloc_data = self._get_res_cls_filters(request_spec)
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'],
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'],
affinity_zone, node = affz_node
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
filtered_nodes &= filtered_rps
if not filtered_nodes:
# if got empty, just break here.
return []
query_filters = {'resources': resources_filter}
filtered_nodes = self.reportclient.\
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.
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',
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'],
self._consume_per_server(context, request_spec, node,
return dest_nodes
filtered_affzs_nodes = self._get_filtered_affzs_nodes(
context, server_group, filtered_nodes,
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]