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, 2 insertions, 1 deletions

diff --git a/openstack/common/rpc/impl_kombu.py b/openstack/common/rpc/impl_kombu.py
index ced4e96..cdbba4e 100644
--- a/openstack/common/rpc/impl_kombu.py
+++ b/openstack/common/rpc/impl_kombu.py
@@ -162,7 +162,8 @@ class ConsumerBase(object):
         def _callback(raw_message):
             message = self.channel.message_to_python(raw_message)
             try:
-                callback(message.payload)
+                msg = rpc_common.deserialize_msg(message.payload)
+                callback(msg)
                 message.ack()
             except Exception:
                 LOG.exception(_("Failed to process message... skipping it."))