-diff -r -N -u3 a/flags.py b/flags.py
---- a/flags.py 2012-08-02 01:34:23.725950885 +0000
-+++ b/flags.py 2012-08-02 01:34:10.914851304 +0000
-@@ -195,6 +195,9 @@
+From 32aae43c94d692d02324c1a299996e5f979f9613 Mon Sep 17 00:00:00 2001
+From: Eugene Kirpichov <ekirpichov@gmail.com>
+Date: Tue, 30 Oct 2012 11:30:59 +0000
+Subject: [PATCH] Support for several HA RabbitMQ servers.
+
+Backport from https://review.openstack.org/#/c/13665/
+
+Change-Id: Ib44abf115b42c3df42771344f6722ce1db043bbd
+---
+ nova/openstack/common/network_utils.py  |   68 ++++++++++++++++++
+ nova/openstack/common/rpc/impl_kombu.py |  120 +++++++++++++++++++++----------
+ openstack-common.conf                   |    2 +-
+ 3 files changed, 150 insertions(+), 40 deletions(-)
+ create mode 100644 nova/openstack/common/network_utils.py
+
+diff --git a/nova/openstack/common/network_utils.py b/nova/openstack/common/network_utils.py
+new file mode 100644
+index 0000000..69f6732
+--- /dev/null
++++ b/nova/openstack/common/network_utils.py
+@@ -0,0 +1,68 @@
++# vim: tabstop=4 shiftwidth=4 softtabstop=4
++
++# Copyright 2012 OpenStack LLC.
++# All Rights Reserved.
++#
++#    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
++#    a copy of the License at
++#
++#         http://www.apache.org/licenses/LICENSE-2.0
++#
++#    Unless required by applicable law or agreed to in writing, software
++#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
++#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
++#    License for the specific language governing permissions and limitations
++#    under the License.
++
++"""
++Network-related utilities and helper functions.
++"""
++
++import logging
++
++LOG = logging.getLogger(__name__)
++
++
++def parse_host_port(address, default_port=None):
++    """
++    Interpret a string as a host:port pair.
++    An IPv6 address MUST be escaped if accompanied by a port,
++    because otherwise ambiguity ensues: 2001:db8:85a3::8a2e:370:7334
++    means both [2001:db8:85a3::8a2e:370:7334] and
++    [2001:db8:85a3::8a2e:370]:7334.
++
++    >>> parse_host_port('server01:80')
++    ('server01', 80)
++    >>> parse_host_port('server01')
++    ('server01', None)
++    >>> parse_host_port('server01', default_port=1234)
++    ('server01', 1234)
++    >>> parse_host_port('[::1]:80')
++    ('::1', 80)
++    >>> parse_host_port('[::1]')
++    ('::1', None)
++    >>> parse_host_port('[::1]', default_port=1234)
++    ('::1', 1234)
++    >>> parse_host_port('2001:db8:85a3::8a2e:370:7334', default_port=1234)
++    ('2001:db8:85a3::8a2e:370:7334', 1234)
++
++    """
++    if address[0] == '[':
++        # Escaped ipv6
++        _host, _port = address[1:].split(']')
++        host = _host
++        if ':' in _port:
++            port = _port.split(':')[1]
++        else:
++            port = default_port
++    else:
++        if address.count(':') == 1:
++            host, port = address.split(':')
++        else:
++            # 0 means ipv4, >1 means ipv6.
++            # We prohibit unescaped ipv6 addresses with port.
++            host = address
++            port = default_port
++
++    return (host, None if port is None else int(port))
+diff --git a/nova/openstack/common/rpc/impl_kombu.py b/nova/openstack/common/rpc/impl_kombu.py
+index fff1ed9..3469a9b 100644
+--- a/nova/openstack/common/rpc/impl_kombu.py
++++ b/nova/openstack/common/rpc/impl_kombu.py
+@@ -33,6 +33,7 @@ from nova.openstack.common import cfg
+ from nova.openstack.common.gettextutils import _
+ from nova.openstack.common.rpc import amqp as rpc_amqp
+ from nova.openstack.common.rpc import common as rpc_common
++from nova.openstack.common import network_utils
+ 
+ kombu_opts = [
+     cfg.StrOpt('kombu_ssl_version',
+@@ -50,10 +51,13 @@ kombu_opts = [
+                      '(valid only if SSL enabled)')),
+     cfg.StrOpt('rabbit_host',
+                default='localhost',
+-               help='the RabbitMQ host'),
++               help='The RabbitMQ broker address where a single node is used'),
      cfg.IntOpt('rabbit_port',
                 default=5672,
-                help='the RabbitMQ port'),
-+    cfg.ListOpt('rabbit_addresses',
-+               default=['$rabbit_host:$rabbit_port'],
-+               help='RabbitMQ HA cluster host:port pairs'),
+-               help='the RabbitMQ port'),
++               help='The RabbitMQ broker port where a single node is used'),
++    cfg.ListOpt('rabbit_hosts',
++                default=['$rabbit_host:$rabbit_port'],
++                help='RabbitMQ HA cluster host:port pairs'),
      cfg.BoolOpt('rabbit_use_ssl',
                  default=False,
                  help='connect over SSL for RabbitMQ'),
-diff -r -N -u3 a/rpc/impl_kombu.py b/rpc/impl_kombu.py
---- a/rpc/impl_kombu.py    2012-08-01 23:18:25.832654796 +0000
-+++ b/rpc/impl_kombu.py    2012-08-01 23:20:56.941282534 +0000
-@@ -77,6 +77,7 @@
-         """Re-declare the queue after a rabbit reconnect"""
-         self.channel = channel
-         self.kwargs['channel'] = channel
-+        self.kwargs['queue_arguments'] = {'x-ha-policy': 'all'}
-         self.queue = kombu.entity.Queue(**self.kwargs)
-         self.queue.declare()
- 
-@@ -348,27 +349,30 @@
+@@ -80,6 +84,11 @@ kombu_opts = [
+     cfg.BoolOpt('rabbit_durable_queues',
+                 default=False,
+                 help='use durable queues in RabbitMQ'),
++    cfg.BoolOpt('rabbit_ha_queues',
++                default=False,
++                help='use H/A queues in RabbitMQ (x-ha-policy: all).'
++                     'You need to wipe RabbitMQ database when '
++                     'changing this option.'),
+ 
+ ]
+ 
+@@ -88,6 +97,20 @@ cfg.CONF.register_opts(kombu_opts)
+ LOG = rpc_common.LOG
+ 
+ 
++def _get_queue_arguments(conf):
++    """Construct the arguments for declaring a queue.
++
++    If the rabbit_ha_queues option is set, we declare a mirrored queue
++    as described here:
++
++      http://www.rabbitmq.com/ha.html
++
++    Setting x-ha-policy to all means that the queue will be mirrored
++    to all nodes in the cluster.
++    """
++    return {'x-ha-policy': 'all'} if conf.rabbit_ha_queues else {}
++
++
+ class ConsumerBase(object):
+     """Consumer base class."""
+ 
+@@ -172,6 +195,7 @@ class DirectConsumer(ConsumerBase):
+         """
+         # Default options
+         options = {'durable': False,
++                   'queue_arguments': _get_queue_arguments(conf),
+                    'auto_delete': True,
+                    'exclusive': True}
+         options.update(kwargs)
+@@ -207,6 +231,7 @@ class TopicConsumer(ConsumerBase):
+         """
+         # Default options
+         options = {'durable': conf.rabbit_durable_queues,
++                   'queue_arguments': _get_queue_arguments(conf),
+                    'auto_delete': False,
+                    'exclusive': False}
+         options.update(kwargs)
+@@ -242,6 +267,7 @@ class FanoutConsumer(ConsumerBase):
+ 
+         # Default options
+         options = {'durable': False,
++                   'queue_arguments': _get_queue_arguments(conf),
+                    'auto_delete': True,
+                    'exclusive': True}
+         options.update(kwargs)
+@@ -289,6 +315,7 @@ class DirectPublisher(Publisher):
+         """
+ 
+         options = {'durable': False,
++                   'queue_arguments': _get_queue_arguments(conf),
+                    'auto_delete': True,
+                    'exclusive': True}
+         options.update(kwargs)
+@@ -304,6 +331,7 @@ class TopicPublisher(Publisher):
+         Kombu options may be passed as keyword args to override defaults
+         """
+         options = {'durable': conf.rabbit_durable_queues,
++                   'queue_arguments': _get_queue_arguments(conf),
+                    'auto_delete': False,
+                    'exclusive': False}
+         options.update(kwargs)
+@@ -319,6 +347,7 @@ class FanoutPublisher(Publisher):
+         Kombu options may be passed as keyword args to override defaults
+         """
+         options = {'durable': False,
++                   'queue_arguments': _get_queue_arguments(conf),
+                    'auto_delete': True,
+                    'exclusive': True}
+         options.update(kwargs)
+@@ -331,6 +360,7 @@ class NotifyPublisher(TopicPublisher):
+ 
+     def __init__(self, conf, channel, topic, **kwargs):
+         self.durable = kwargs.pop('durable', conf.rabbit_durable_queues)
++        self.queue_arguments = _get_queue_arguments(conf)
+         super(NotifyPublisher, self).__init__(conf, channel, topic, **kwargs)
+ 
+     def reconnect(self, channel):
+@@ -343,7 +373,8 @@ class NotifyPublisher(TopicPublisher):
+                                    exchange=self.exchange,
+                                    durable=self.durable,
+                                    name=self.routing_key,
+-                                   routing_key=self.routing_key)
++                                   routing_key=self.routing_key,
++                                   queue_arguments=self.queue_arguments)
+         queue.declare()
+ 
+ 
+@@ -368,31 +399,39 @@ class Connection(object):
+ 
+         if server_params is None:
+             server_params = {}
+-
          # Keys to translate from server_params to kombu params
          server_params_to_kombu_params = {'username': 'userid'}
  

+                self._connect(params)
                 return
-            except (self.connection_errors, IOError), e:
+            except (self.connection_errors, self.channel_errors, IOError), e:
                 pass
             except Exception, e:
                 # NOTE(comstud): Unfortunately it's possible for amqplib
@@ -460,7 +463,7 @@
             log_info = {}
             log_info['err_str'] = str(e)
             log_info['max_retries'] = self.max_retries

         """
         if self.connection:
             LOG.info(_("Reconnecting to AMQP server on "
-                     "%(hostname)s:%(port)d") % self.params)
+                     "%(hostname)s:%(port)d") % params)
             try:
                 self.connection.close()
-            except self.connection_errors:
+            except (self.connection_errors, self.channel_errors):
                 pass
             # Setting this in case the next statement fails, though
             # it shouldn't be doing any network operations, yet.
             self.connection = None
-        self.connection = kombu.connection.BrokerConnection(**self.params)
+        self.connection = kombu.connection.BrokerConnection(**params)
         self.connection_errors = self.connection.connection_errors
+        self.channel_errors = self.connection.channel_errors
         if self.memory_transport:
             # Kludge to speed up tests.
             self.connection.transport.polling_interval = 0.0
@@ -450,8 +490,8 @@ class Connection(object):
             self.channel._new_queue('ae.undeliver')
         for consumer in self.consumers:
             consumer.reconnect(self.channel)
-        LOG.info(_('Connected to AMQP server on %(hostname)s:%(port)d'),
-                 self.params)
+        LOG.info(_('Connected to AMQP server on %(hostname)s:%(port)d') %
+                 params)
 
     def reconnect(self):
         """Handles reconnecting and re-establishing queues.
@@ -464,11 +504,12 @@ class Connection(object):
 
         attempt = 0
         while True:
