Port to oslo.messaging

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

1 files changed, 141 insertions, 0 deletions

diff --git a/nova/rpc.py b/nova/rpc.py
new file mode 100644
index 0000000000..609a2769ae
--- /dev/null
+++ b/nova/rpc.py
@@ -0,0 +1,141 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2013 Red Hat, Inc.
+#
+#    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.
+
+__all__ = [
+    'init',
+    'cleanup',
+    'set_defaults',
+    'add_extra_exmods',
+    'clear_extra_exmods',
+    'get_allowed_exmods',
+    'RequestContextSerializer',
+    'get_client',
+    'get_server',
+    'get_notifier',
+    'TRANSPORT_ALIASES',
+]
+
+from oslo.config import cfg
+from oslo import messaging
+
+import nova.context
+import nova.exception
+
+CONF = cfg.CONF
+TRANSPORT = None
+NOTIFIER = None
+
+ALLOWED_EXMODS = [
+    nova.exception.__name__,
+]
+EXTRA_EXMODS = []
+
+# NOTE(markmc): The nova.openstack.common.rpc entries are for backwards compat
+# with Havana rpc_backend configuration values. The nova.rpc entries are for
+# compat with Essex values.
+TRANSPORT_ALIASES = {
+    'nova.openstack.common.rpc.impl_kombu': 'rabbit',
+    'nova.openstack.common.rpc.impl_qpid': 'qpid',
+    'nova.openstack.common.rpc.impl_zmq': 'zmq',
+    'nova.rpc.impl_kombu': 'rabbit',
+    'nova.rpc.impl_qpid': 'qpid',
+    'nova.rpc.impl_zmq': 'zmq',
+}
+
+
+def init(conf):
+    global TRANSPORT, NOTIFIER
+    exmods = get_allowed_exmods()
+    TRANSPORT = messaging.get_transport(conf,
+                                        allowed_remote_exmods=exmods,
+                                        aliases=TRANSPORT_ALIASES)
+    NOTIFIER = messaging.Notifier(TRANSPORT)
+
+
+def cleanup():
+    global TRANSPORT, NOTIFIER
+    assert TRANSPORT is not None
+    assert NOTIFIER is not None
+    TRANSPORT.cleanup()
+    TRANSPORT = NOTIFIER = None
+
+
+def set_defaults(control_exchange):
+    messaging.set_transport_defaults(control_exchange)
+
+
+def add_extra_exmods(*args):
+    EXTRA_EXMODS.extend(args)
+
+
+def clear_extra_exmods():
+    del EXTRA_EXMODS[:]
+
+
+def get_allowed_exmods():
+    return ALLOWED_EXMODS + EXTRA_EXMODS
+
+
+class RequestContextSerializer(messaging.Serializer):
+
+    def __init__(self, base):
+        self._base = base
+
+    def serialize_entity(self, context, entity):
+        if not self._base:
+            return entity
+        return self._base.serialize_entity(context, entity)
+
+    def deserialize_entity(self, context, entity):
+        if not self._base:
+            return entity
+        return self._base.deserialize_entity(context, entity)
+
+    def serialize_context(self, context):
+        return context.to_dict()
+
+    def deserialize_context(self, context):
+        return nova.context.RequestContext.from_dict(context)
+
+
+def get_transport_url(url_str=None):
+    return messaging.TransportURL.parse(CONF, url_str, TRANSPORT_ALIASES)
+
+
+def get_client(target, version_cap=None, serializer=None):
+    assert TRANSPORT is not None
+    serializer = RequestContextSerializer(serializer)
+    return messaging.RPCClient(TRANSPORT,
+                               target,
+                               version_cap=version_cap,
+                               serializer=serializer)
+
+
+def get_server(target, endpoints, serializer=None):
+    assert TRANSPORT is not None
+    serializer = RequestContextSerializer(serializer)
+    return messaging.get_rpc_server(TRANSPORT,
+                                    target,
+                                    endpoints,
+                                    executor='eventlet',
+                                    serializer=serializer)
+
+
+def get_notifier(service=None, host=None, publisher_id=None):
+    assert NOTIFIER is not None
+    if not publisher_id:
+        publisher_id = "%s.%s" % (service, host or CONF.host)
+    return NOTIFIER.prepare(publisher_id=publisher_id)