1a91aacb85
The oslo.messaging library takes the existing RPC code from oslo and wraps it in a sane API with well defined semantics around which we can make a commitment to retain compatibility in future. The patch is large, but the changes can be summarized as: * oslo.messaging>=1.3.0a4 is required; a proper 1.3.0 release will be pushed before the icehouse release candidates. * The new rpc module has init() and cleanup() methods which manage the global oslo.messaging transport state. The TRANSPORT and NOTIFIER globals are conceptually similar to the current RPCIMPL global, except we're free to create and use alternate Transport objects in e.g. the cells code. * The rpc.get_{client,server,notifier}() methods are just helpers which wrap the global messaging state, specifiy serializers and specify the use of the eventlet executor. * In oslo.messaging, a request context is expected to be a dict so we add a RequestContextSerializer which can serialize to and from dicts using RequestContext.{to,from}_dict() * The allowed_rpc_exception_modules configuration option is replaced by an allowed_remote_exmods get_transport() parameter. This is not something that users ever need to configure, but it is something each project using oslo.messaging needs to be able to customize. * The nova.rpcclient module is removed; it was only a helper class to allow us split a lot of the more tedious changes out of this patch. * Finalizing the port from RpcProxy to RPCClient is straightforward. We put the default topic, version and namespace into a Target and contstruct the client using that. * Porting endpoint classes (like ComputeManager) just involves setting a target attribute on the class. * The @client_exceptions() decorator has been renamed to @expected_exceptions since it's used on the server side to designate exceptions we expect the decorated method to raise. * We maintain a global NOTIFIER object and create specializations of it with specific publisher IDs in order to avoid notification driver loading overhead. * rpc.py contains transport aliases for backwards compatibility purposes. setup.cfg also contains notification driver aliases for backwards compat. * The messaging options are moved about in nova.conf.sample because the options are advertised via a oslo.config.opts entry point and picked up by the generator. * We use messaging.ConfFixture in tests to override oslo.messaging config options, rather than making assumptions about the options registered by the library. The porting of cells code is particularly tricky: * messaging.TransportURL parse() and str() replaces the [un]parse_transport_url() methods. Note the complication that an oslo.messaging transport URL can actually have multiple hosts in order to support message broker clustering. Also the complication of transport aliases in rpc.get_transport_url(). * proxy_rpc_to_manager() is fairly nasty. Right now, we're proxying the on-the-wire message format over this call, but you can't supply such messages to oslo.messaging's cast()/call() methods. Rather than change the inter-cell RPC API to suit oslo.messaging, we instead just unpack the topic, server, method and args from the message on the remote side. cells_api.RPCClientCellsProxy is a mock RPCClient implementation which allows us to wrap up a RPC in the message format currently used for inter-cell RPCs. * Similarly, proxy_rpc_to_manager uses the on-the-wire format for exception serialization, but this format is an implementation detail of oslo.messaging's transport drivers. So, we need to duplicate the exception serialization code in cells.messaging. We may find a way to reconcile this in future - for example a ExceptionSerializer class might work, but with the current format it might be difficult for the deserializer to generically detect a serialized exception. * CellsRPCDriver.start_servers() and InterCellRPCAPI._get_client() need close review, but they're pretty straightforward ports of code to listen on some specialized topics and connect to a remote cell using its transport URL. blueprint: oslo-messaging Change-Id: Ib613e6300f2c215be90f924afbd223a3da053a69
170 lines
6.4 KiB
Python
170 lines
6.4 KiB
Python
# 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.
|
|
|
|
"""Utility methods for scheduling."""
|
|
|
|
import sys
|
|
|
|
from nova.compute import flavors
|
|
from nova.compute import utils as compute_utils
|
|
from nova import db
|
|
from nova import notifications
|
|
from nova.objects import base as obj_base
|
|
from nova.objects import instance as instance_obj
|
|
from nova.openstack.common.gettextutils import _
|
|
from nova.openstack.common import jsonutils
|
|
from nova.openstack.common import log as logging
|
|
from nova import rpc
|
|
|
|
LOG = logging.getLogger(__name__)
|
|
|
|
|
|
def build_request_spec(ctxt, image, instances, instance_type=None):
|
|
"""Build a request_spec for the scheduler.
|
|
|
|
The request_spec assumes that all instances to be scheduled are the same
|
|
type.
|
|
"""
|
|
instance = instances[0]
|
|
if isinstance(instance, instance_obj.Instance):
|
|
instance = obj_base.obj_to_primitive(instance)
|
|
|
|
if instance_type is None:
|
|
instance_type = flavors.extract_flavor(instance)
|
|
# NOTE(comstud): This is a bit ugly, but will get cleaned up when
|
|
# we're passing an InstanceType internal object.
|
|
extra_specs = db.flavor_extra_specs_get(ctxt, instance_type['flavorid'])
|
|
instance_type['extra_specs'] = extra_specs
|
|
request_spec = {
|
|
'image': image or {},
|
|
'instance_properties': instance,
|
|
'instance_type': instance_type,
|
|
'num_instances': len(instances),
|
|
# NOTE(alaski): This should be removed as logic moves from the
|
|
# scheduler to conductor. Provides backwards compatibility now.
|
|
'instance_uuids': [inst['uuid'] for inst in instances]}
|
|
return jsonutils.to_primitive(request_spec)
|
|
|
|
|
|
def set_vm_state_and_notify(context, service, method, updates, ex,
|
|
request_spec, db):
|
|
"""changes VM state and notifies."""
|
|
LOG.warning(_("Failed to %(service)s_%(method)s: %(ex)s"),
|
|
{'service': service, 'method': method, 'ex': ex})
|
|
|
|
vm_state = updates['vm_state']
|
|
properties = request_spec.get('instance_properties', {})
|
|
# NOTE(vish): We shouldn't get here unless we have a catastrophic
|
|
# failure, so just set all instances to error. if uuid
|
|
# is not set, instance_uuids will be set to [None], this
|
|
# is solely to preserve existing behavior and can
|
|
# be removed along with the 'if instance_uuid:' if we can
|
|
# verify that uuid is always set.
|
|
uuids = [properties.get('uuid')]
|
|
from nova.conductor import api as conductor_api
|
|
conductor = conductor_api.LocalAPI()
|
|
notifier = rpc.get_notifier(service)
|
|
for instance_uuid in request_spec.get('instance_uuids') or uuids:
|
|
if instance_uuid:
|
|
state = vm_state.upper()
|
|
LOG.warning(_('Setting instance to %s state.'), state,
|
|
instance_uuid=instance_uuid)
|
|
|
|
# update instance state and notify on the transition
|
|
(old_ref, new_ref) = db.instance_update_and_get_original(
|
|
context, instance_uuid, updates)
|
|
notifications.send_update(context, old_ref, new_ref,
|
|
service=service)
|
|
compute_utils.add_instance_fault_from_exc(context,
|
|
conductor,
|
|
new_ref, ex, sys.exc_info())
|
|
|
|
payload = dict(request_spec=request_spec,
|
|
instance_properties=properties,
|
|
instance_id=instance_uuid,
|
|
state=vm_state,
|
|
method=method,
|
|
reason=ex)
|
|
|
|
event_type = '%s.%s' % (service, method)
|
|
notifier.error(context, event_type, payload)
|
|
|
|
|
|
def populate_filter_properties(filter_properties, host_state):
|
|
"""Add additional information to the filter properties after a node has
|
|
been selected by the scheduling process.
|
|
"""
|
|
if isinstance(host_state, dict):
|
|
host = host_state['host']
|
|
nodename = host_state['nodename']
|
|
limits = host_state['limits']
|
|
else:
|
|
host = host_state.host
|
|
nodename = host_state.nodename
|
|
limits = host_state.limits
|
|
|
|
# Adds a retry entry for the selected compute host and node:
|
|
_add_retry_host(filter_properties, host, nodename)
|
|
|
|
# Adds oversubscription policy
|
|
if not filter_properties.get('force_hosts'):
|
|
filter_properties['limits'] = limits
|
|
|
|
|
|
def _add_retry_host(filter_properties, host, node):
|
|
"""Add a retry entry for the selected compute node. In the event that
|
|
the request gets re-scheduled, this entry will signal that the given
|
|
node has already been tried.
|
|
"""
|
|
retry = filter_properties.get('retry', None)
|
|
force_hosts = filter_properties.get('force_hosts', [])
|
|
force_nodes = filter_properties.get('force_nodes', [])
|
|
if not retry or force_hosts or force_nodes:
|
|
return
|
|
hosts = retry['hosts']
|
|
hosts.append([host, node])
|
|
|
|
|
|
def parse_options(opts, sep='=', converter=str, name=""):
|
|
"""Parse a list of options, each in the format of <key><sep><value>. Also
|
|
use the converter to convert the value into desired type.
|
|
|
|
:params opts: list of options, e.g. from oslo.config.cfg.ListOpt
|
|
:params sep: the separator
|
|
:params converter: callable object to convert the value, should raise
|
|
ValueError for conversion failure
|
|
:params name: name of the option
|
|
|
|
:returns: a lists of tuple of values (key, converted_value)
|
|
"""
|
|
good = []
|
|
bad = []
|
|
for opt in opts:
|
|
try:
|
|
key, seen_sep, value = opt.partition(sep)
|
|
value = converter(value)
|
|
except ValueError:
|
|
key = None
|
|
value = None
|
|
if key and seen_sep and value is not None:
|
|
good.append((key, value))
|
|
else:
|
|
bad.append(opt)
|
|
if bad:
|
|
LOG.warn(_("Ignoring the invalid elements of the option "
|
|
"%(name)s: %(options)s"),
|
|
{'name': name,
|
|
'options': ", ".join(bad)})
|
|
return good
|