Sync RPC module from oslo

This change syncs the following patches from stable/havana oslo RPC:
- I67923cb024bbd143edc8edccf35b9b400df31eb3
- Ia148baa6e1ec632789ac3621c85173c2c16f3918

RabbitMQ: advance thru the list of brokers on reconnect

In RabbitMQ implementation, when using multiple rabbit_hosts, we don't want to
immediately retry failed connection for the same failed broker. This was not
the case in existing implementation though, where we've always attempted to
reconnect starting from the first broker in the list of candidates. So if the
first broker failed, we initiated reconnect to the same failed broker.

This change makes reconnect() implementation to select the next broker in the
list. This also means that non-failure reconnect attempts will also switch the
current broker. All in all, users should not rely on any particular order to
use brokers from the list, so this should not constitute an issue.

Qpid: advance thru the list of brokers on reconnect

In Qpid implementation, when using multiple qpid_hosts, we don't want to
immediately retry failed connection for the same failed broker. This was not
the case in existing implementation though, where we've always attempted to
reconnect starting from the first broker in the list of candidates. So if the
first broker failed, we initiated reconnect to the same failed broker.

This change makes reconnect() implementation to select the next broker in the
list. This also means that non-failure reconnect attempts will also switch the
current broker. All in all, users should not rely on any particular order to
use brokers from the list, so this should not constitute an issue.

Partial-Bug: #1261631
Change-Id: Icde6b3a1fa173be2a3ccc729b7ed739dd4f4aa11

2 files changed, 9 insertions, 4 deletions

diff --git a/ceilometer/openstack/common/rpc/impl_kombu.py b/ceilometer/openstack/common/rpc/impl_kombu.py
index cf55a9f0..36355322 100644
--- a/ceilometer/openstack/common/rpc/impl_kombu.py
+++ b/ceilometer/openstack/common/rpc/impl_kombu.py
@@ -459,6 +459,9 @@ class Connection(object):
 
         self.params_list = params_list
 
+        brokers_count = len(self.params_list)
+        self.next_broker_indices = itertools.cycle(range(brokers_count))
+
         self.memory_transport = self.conf.fake_rabbit
 
         self.connection = None
@@ -529,7 +532,7 @@ class Connection(object):
 
         attempt = 0
         while True:
-            params = self.params_list[attempt % len(self.params_list)]
+            params = self.params_list[next(self.next_broker_indices)]
             attempt += 1
             try:
                 self._connect(params)
diff --git a/ceilometer/openstack/common/rpc/impl_qpid.py b/ceilometer/openstack/common/rpc/impl_qpid.py
index a697d911..9e5d8916 100644
--- a/ceilometer/openstack/common/rpc/impl_qpid.py
+++ b/ceilometer/openstack/common/rpc/impl_qpid.py
@@ -468,6 +468,10 @@ class Connection(object):
         self.brokers = params['qpid_hosts']
         self.username = params['username']
         self.password = params['password']
+
+        brokers_count = len(self.brokers)
+        self.next_broker_indices = itertools.cycle(range(brokers_count))
+
         self.connection_create(self.brokers[0])
         self.reconnect()
 
@@ -495,7 +499,6 @@ class Connection(object):
 
     def reconnect(self):
         """Handles reconnecting and re-establishing sessions and queues."""
-        attempt = 0
         delay = 1
         while True:
             # Close the session if necessary
@@ -505,8 +508,7 @@ class Connection(object):
                 except qpid_exceptions.ConnectionError:
                     pass
 
-            broker = self.brokers[attempt % len(self.brokers)]
-            attempt += 1
+            broker = self.brokers[next(self.next_broker_indices)]
 
             try:
                 self.connection_create(broker)