Support capping message versions in the client.

When doing a rolling upgrade, we need to be able to tell all rpc clients
to hold off on sending newer versions of messages until all nodes
understand the new message version.  This patch adds the oslo component
of this.

It's quite simple.  The rpc proxy just stores the version cap and will
raise an exception if code ever tries to send a message that exceeds
this cap.

Allowing the cap to be configured and generating different types of
messages based on the configured value is the hard part here, but that
is left up to the project using the rpc library.

Implements blueprint rpc-version-control.

Change-Id: Ia69db03a80dc3b1c63d52c7e163ea3cfec80c882

2 files changed, 15 insertions, 3 deletions

diff --git a/openstack/common/rpc/common.py b/openstack/common/rpc/common.py
index d74425d..5a7e525 100644
--- a/openstack/common/rpc/common.py
+++ b/openstack/common/rpc/common.py
@@ -158,6 +158,10 @@ class UnsupportedRpcEnvelopeVersion(RPCException):
                 "not supported by this endpoint.")
 
 
+class RpcVersionCapError(RPCException):
+    message = _("Specified RPC version cap, %(version_cap)s, is too low")
+
+
 class Connection(object):
     """A connection, returned by rpc.create_connection().
 
diff --git a/openstack/common/rpc/proxy.py b/openstack/common/rpc/proxy.py
index 284f375..0b311de 100644
--- a/openstack/common/rpc/proxy.py
+++ b/openstack/common/rpc/proxy.py
@@ -1,6 +1,6 @@
 # vim: tabstop=4 shiftwidth=4 softtabstop=4
 
-# Copyright 2012 Red Hat, Inc.
+# Copyright 2012-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
@@ -23,6 +23,7 @@ For more information about rpc API version numbers, see:
 
 
 from openstack.common import rpc
+from openstack.common.rpc import common as rpc_common
 
 
 class RpcProxy(object):
@@ -34,16 +35,19 @@ class RpcProxy(object):
     rpc API.
     """
 
-    def __init__(self, topic, default_version):
+    def __init__(self, topic, default_version, version_cap=None):
         """Initialize an RpcProxy.
 
         :param topic: The topic to use for all messages.
         :param default_version: The default API version to request in all
                outgoing messages.  This can be overridden on a per-message
                basis.
+        :param version_cap: Optionally cap the maximum version used for sent
+               messages.
         """
         self.topic = topic
         self.default_version = default_version
+        self.version_cap = version_cap
         super(RpcProxy, self).__init__()
 
     def _set_version(self, msg, vers):
@@ -52,7 +56,11 @@ class RpcProxy(object):
         :param msg: The message having a version added to it.
         :param vers: The version number to add to the message.
         """
-        msg['version'] = vers if vers else self.default_version
+        v = vers if vers else self.default_version
+        if (self.version_cap and not
+                rpc_common.version_is_compatible(self.version_cap, v)):
+            raise rpc_common.RpcVersionCapError(version=self.version_cap)
+        msg['version'] = v
 
     def _get_topic(self, topic):
         """Return the topic to use for a message."""