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

2 files changed, 36 insertions, 7 deletions

diff --git a/tests/unit/rpc/test_common.py b/tests/unit/rpc/test_common.py
index 6f8b7ff..78f4b93 100644
--- a/tests/unit/rpc/test_common.py
+++ b/tests/unit/rpc/test_common.py
@@ -236,3 +236,32 @@ class RpcCommonTestCase(test_utils.BaseTestCase):
 
         self.assertRaises(rpc_common.ClientException, naughty)
         self.assertRaises(ValueError, really_naughty)
+
+    def test_serialize_msg_v1(self):
+        self.stubs.Set(rpc_common, '_SEND_RPC_ENVELOPE', False)
+        msg = {'foo': 'bar'}
+        self.assertEqual(msg, rpc_common.serialize_msg(msg))
+
+    def test_serialize_msg_v2(self):
+        self.stubs.Set(rpc_common, '_SEND_RPC_ENVELOPE', True)
+        msg = {'foo': 'bar'}
+        s_msg = {'oslo.version': rpc_common._RPC_ENVELOPE_VERSION,
+                 'oslo.message': jsonutils.dumps(msg)}
+        serialized = rpc_common.serialize_msg(msg)
+
+        self.assertEqual(s_msg, rpc_common.serialize_msg(msg))
+
+        self.assertEqual(msg, rpc_common.deserialize_msg(serialized))
+
+    def test_deserialize_msg_no_envelope(self):
+        self.assertEqual(1, rpc_common.deserialize_msg(1))
+        self.assertEqual([], rpc_common.deserialize_msg([]))
+        self.assertEqual({}, rpc_common.deserialize_msg({}))
+        self.assertEqual('foo', rpc_common.deserialize_msg('foo'))
+
+    def test_deserialize_msg_bad_version(self):
+        s_msg = {'oslo.version': '8675309.0',
+                 'oslo.message': 'whatever'}
+
+        self.assertRaises(rpc_common.UnsupportedRpcEnvelopeVersion,
+                          rpc_common.deserialize_msg, s_msg)
diff --git a/tests/unit/rpc/test_kombu.py b/tests/unit/rpc/test_kombu.py
index 5da05de..921415d 100644
--- a/tests/unit/rpc/test_kombu.py
+++ b/tests/unit/rpc/test_kombu.py
@@ -118,7 +118,7 @@ class RpcKombuTestCase(common.BaseRpcAMQPTestCase):
             self.received_message = message
 
         conn.declare_topic_consumer('a_topic', _callback)
-        conn.topic_send('a_topic', message)
+        conn.topic_send('a_topic', rpc_common.serialize_msg(message))
         conn.consume(limit=1)
         conn.close()
 
@@ -138,7 +138,7 @@ class RpcKombuTestCase(common.BaseRpcAMQPTestCase):
 
         conn.declare_topic_consumer('a_topic', _callback,
                                     exchange_name="foorbar")
-        conn.topic_send('a_topic', message)
+        conn.topic_send('a_topic', rpc_common.serialize_msg(message))
         conn.consume(limit=1)
         conn.close()
 
@@ -162,7 +162,7 @@ class RpcKombuTestCase(common.BaseRpcAMQPTestCase):
 
         conn.declare_topic_consumer('a_topic', _callback1, queue_name='queue1')
         conn.declare_topic_consumer('a_topic', _callback2, queue_name='queue2')
-        conn.topic_send('a_topic', message)
+        conn.topic_send('a_topic', rpc_common.serialize_msg(message))
         conn.consume(limit=2)
         conn.close()
 
@@ -192,7 +192,7 @@ class RpcKombuTestCase(common.BaseRpcAMQPTestCase):
                                     exchange_name="abc")
         conn.declare_topic_consumer('a_topic', _callback2, queue_name='queue2',
                                     exchange_name="abc")
-        conn.topic_send('a_topic', message)
+        conn.topic_send('a_topic', rpc_common.serialize_msg(message))
         conn.consume(limit=2)
         conn.close()
 
@@ -222,7 +222,7 @@ class RpcKombuTestCase(common.BaseRpcAMQPTestCase):
                                     exchange_name="abc")
         conn.declare_topic_consumer('a_topic', _callback2, queue_name='queue2',
                                     exchange_name="def")
-        conn.topic_send('a_topic', message)
+        conn.topic_send('a_topic', rpc_common.serialize_msg(message))
         conn.consume(limit=2)
         conn.close()
 
@@ -241,7 +241,7 @@ class RpcKombuTestCase(common.BaseRpcAMQPTestCase):
             self.received_message = message
 
         conn.declare_direct_consumer('a_direct', _callback)
-        conn.direct_send('a_direct', message)
+        conn.direct_send('a_direct', rpc_common.serialize_msg(message))
         conn.consume(limit=1)
         conn.close()
 
@@ -438,7 +438,7 @@ class RpcKombuTestCase(common.BaseRpcAMQPTestCase):
             self.received_message = message
 
         conn.declare_direct_consumer('a_direct', _callback)
-        conn.direct_send('a_direct', message)
+        conn.direct_send('a_direct', rpc_common.serialize_msg(message))
 
         info = _raise_exc_stub(self.stubs, 1, conn.connection,
                                'drain_events', 'foo timeout foo')