Add a rpc envelope format version number.

This patch adds a message envelope that includes a envelope format
version number.  This message envelope will allow us to embed
additional metadata later on, such as a signature for the message
payload.

Up to this point, we've deferred message serialization as a
responsibility of the messaging library we're using by passing it a
message as Python types and letting it deal with how to pass it over a
network.  This patch adds json serialization in the rpc layer of the
application message payload before passing the message down into the
messaging library.

There are some benefits to be gained by doing a pass at serialization
ourselves.  As an example, we occasionally hit serialization bugs that
only affect some messaging drivers.  The kombu driver has always had a
nice advantage.  It uses anyjson internally for serializing messages,
which we hook into using our jsonutils module.  When there is a problem
serializing, we automatically use to_primitive() to fix it.  This patch
allows all drivers to take advantage of this automatic message fix-up.

This also creates a convenient common hook point for messages coming in
and out of the system, regardless of the driver in use.

While this changes the base format of the messages sent between nodes,
it has been done in a backwards compatible manner.  The new message
format will not be used by default.  The idea is that all nodes will be
upgraded to a version that is capable of receiving the new format
(Grizzly) before switching it on.  We will turn it on post-Grizzly.

Implement blueprint version-rpc-messages.

Change-Id: Ib6b2d11ca42abaa64c40986d72233e7048e504a0

1 files changed, 118 insertions, 0 deletions

diff --git a/openstack/common/rpc/common.py b/openstack/common/rpc/common.py
index efdf26f..a3a3699 100644
--- a/openstack/common/rpc/common.py
+++ b/openstack/common/rpc/common.py
@@ -21,6 +21,7 @@ import copy
 import sys
 import traceback
 
+from openstack.common import cfg
 from openstack.common.gettextutils import _
 from openstack.common import importutils
 from openstack.common import jsonutils
@@ -28,9 +29,50 @@ from openstack.common import local
 from openstack.common import log as logging
 
 
+CONF = cfg.CONF
 LOG = logging.getLogger(__name__)
 
 
+'''RPC Envelope Version.
+
+This version number applies to the top level structure of messages sent out.
+It does *not* apply to the message payload, which must be versioned
+independently.  For example, when using rpc APIs, a version number is applied
+for changes to the API being exposed over rpc.  This version number is handled
+in the rpc proxy and dispatcher modules.
+
+This version number applies to the message envelope that is used in the
+serialization done inside the rpc layer.  See serialize_msg() and
+deserialize_msg().
+
+The current message format (version 2.0) is very simple.  It is:
+
+    {
+        'oslo.version': <RPC Envelope Version as a String>,
+        'oslo.message': <Application Message Payload, JSON encoded>
+    }
+
+Message format version '1.0' is just considered to be the messages we sent
+without a message envelope.
+
+So, the current message envelope just includes the envelope version.  It may
+eventually contain additional information, such as a signature for the message
+payload.
+
+We will JSON encode the application message payload.  The message envelope,
+which includes the JSON encoded application message body, will be passed down
+to the messaging libraries as a dict.
+'''
+_RPC_ENVELOPE_VERSION = '2.0'
+
+_VERSION_KEY = 'oslo.version'
+_MESSAGE_KEY = 'oslo.message'
+
+
+# TODO(russellb) Turn this on after Grizzly.
+_SEND_RPC_ENVELOPE = False
+
+
 class RPCException(Exception):
     message = _("An unknown RPC related exception occurred.")
 
@@ -91,6 +133,11 @@ class UnsupportedRpcVersion(RPCException):
                 "this endpoint.")
 
 
+class UnsupportedRpcEnvelopeVersion(RPCException):
+    message = _("Specified RPC envelope version, %(version)s, "
+                "not supported by this endpoint.")
+
+
 class Connection(object):
     """A connection, returned by rpc.create_connection().
 
@@ -344,3 +391,74 @@ def client_exceptions(*exceptions):
             return catch_client_exception(exceptions, func, *args, **kwargs)
         return inner
     return outer
+
+
+def version_is_compatible(imp_version, version):
+    """Determine whether versions are compatible.
+
+    :param imp_version: The version implemented
+    :param version: The version requested by an incoming message.
+    """
+    version_parts = version.split('.')
+    imp_version_parts = imp_version.split('.')
+    if int(version_parts[0]) != int(imp_version_parts[0]):  # Major
+        return False
+    if int(version_parts[1]) > int(imp_version_parts[1]):  # Minor
+        return False
+    return True
+
+
+def serialize_msg(raw_msg):
+    if not _SEND_RPC_ENVELOPE:
+        return raw_msg
+
+    # NOTE(russellb) See the docstring for _RPC_ENVELOPE_VERSION for more
+    # information about this format.
+    msg = {_VERSION_KEY: _RPC_ENVELOPE_VERSION,
+           _MESSAGE_KEY: jsonutils.dumps(raw_msg)}
+
+    return msg
+
+
+def deserialize_msg(msg):
+    # NOTE(russellb): Hang on to your hats, this road is about to
+    # get a little bumpy.
+    #
+    # Robustness Principle:
+    #    "Be strict in what you send, liberal in what you accept."
+    #
+    # At this point we have to do a bit of guessing about what it
+    # is we just received.  Here is the set of possibilities:
+    #
+    # 1) We received a dict.  This could be 2 things:
+    #
+    #   a) Inspect it to see if it looks like a standard message envelope.
+    #      If so, great!
+    #
+    #   b) If it doesn't look like a standard message envelope, it could either
+    #      be a notification (which we don't wrap), or a message from before
+    #      we added a message envelope (referred to as version 1.0).
+    #      Just return the message as-is.
+    #
+    # 2) It's any other non-dict type.  Just return it and hope for the best.
+    #    This case covers return values from rpc.call() from before message
+    #    envelopes were used.  (messages to call a method were always a dict)
+
+    if not isinstance(msg, dict):
+        # See #2 above.
+        return msg
+
+    base_envelope_keys = (_VERSION_KEY, _MESSAGE_KEY)
+    if not all(map(lambda key: key in msg, base_envelope_keys)):
+        #  See #1.b above.
+        return msg
+
+    # At this point we think we have the message envelope
+    # format we were expecting. (#1.a above)
+
+    if not version_is_compatible(_RPC_ENVELOPE_VERSION, msg[_VERSION_KEY]):
+        raise UnsupportedRpcEnvelopeVersion(version=msg[_VERSION_KEY])
+
+    raw_msg = jsonutils.loads(msg[_MESSAGE_KEY])
+
+    return raw_msg