<filename>mogan/scheduler/filter_scheduler.py
# 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]

<filename>mogan/scheduler/filter_scheduler.py
# 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]

# 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. Scheduler that can be used for filtering and weighing. Log requests with exceptions from previous server operations. # string-ified exception from server # no exception info from a previous attempt, skip # no previously attempted nodes, skip Populate filter properties with history of retries for request.          If maximum retries is exceeded, raise NoValidNode. # re-scheduling is disabled. # Add availability_zone aggregate # if no aggregates match with the key/value, # fail the scheduling. 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)] # None indicates no matching aggregates # if got empty, just break here. # initialize the filtered_nodes # if got empty, just break here. # 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.