Add support for Qpid to nova.rpc.
Implements blueprint apache-qpid-rpc.
This patch adds a new implementation of the nova.rpc API that uses Qpid
as the messaging backend. We first considered trying to add Qpid
support to kombu, but the kombu API did not map very well to Qpid, which
is based on a newer version of AMQP. It was easier to just map the nova
rpc API to Qpid.
For more information about how to use Qpid with Nova, see this wiki
page:
http://wiki.openstack.org/QpidSupport
The structure of impl_qpid is largely based on impl_kombu, but adapted
to use the Qpid APIs as necessary. This patch also factors out some of
the code shared with impl_kombu into nova.rpc.common to try to cut down
on duplicated code.
The unit tests were written from scratch instead of taking advantage
of the common rpc unit tests. The common unit tests only work if qpidd
is running, but they do pass if enabled. The unit tests for impl_qpid
instead use mox to mock out the Qpid objects and ensure the right
operations happen on them when the rpc API is exercised.
This patch was a joint effort between myself and William Henry, which is
why he was added to the Authors file in this patch.
Change-Id: Ibacaa956e016ef96f014443074e2a4622e31f090
This commit is contained in:
Russell Bryant
1
Authors
1
Authors
@@ -151,6 +151,7 @@ Vasiliy Shlykov <vash@vasiliyshlykov.org>
|
||||
Vishvananda Ishaya <vishvananda@gmail.com>
|
||||
Vivek Y S <vivek.ys@gmail.com>
|
||||
Vladimir Popovski <vladimir@zadarastorage.com>
|
||||
William Henry <whenry@redhat.com>
|
||||
William Kelly <william.kelly@rackspace.com>
|
||||
William Wolf <throughnothing@gmail.com>
|
||||
Yoshiaki Tamura <yoshi@midokura.jp>
|
||||
|
||||
355
nova/rpc/amqp.py
Normal file
355
nova/rpc/amqp.py
Normal file
@@ -0,0 +1,355 @@
|
||||
# vim: tabstop=4 shiftwidth=4 softtabstop=4
|
||||
|
||||
# Copyright 2010 United States Government as represented by the
|
||||
# Administrator of the National Aeronautics and Space Administration.
|
||||
# All Rights Reserved.
|
||||
# Copyright 2011 - 2012, 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
|
||||
# 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.
|
||||
|
||||
"""
|
||||
Shared code between AMQP based nova.rpc implementations.
|
||||
|
||||
The code in this module is shared between the rpc implemenations based on AMQP.
|
||||
Specifically, this includes impl_kombu and impl_qpid. impl_carrot also uses
|
||||
AMQP, but is deprecated and predates this code.
|
||||
"""
|
||||
|
||||
import inspect
|
||||
import sys
|
||||
import traceback
|
||||
import uuid
|
||||
|
||||
from eventlet import greenpool
|
||||
from eventlet import pools
|
||||
|
||||
from nova import context
|
||||
from nova import exception
|
||||
from nova import flags
|
||||
from nova import local
|
||||
import nova.rpc.common as rpc_common
|
||||
from nova.rpc.common import LOG
|
||||
|
||||
|
||||
FLAGS = flags.FLAGS
|
||||
|
||||
|
||||
ConnectionClass = None
|
||||
|
||||
|
||||
class Pool(pools.Pool):
|
||||
"""Class that implements a Pool of Connections."""
|
||||
def __init__(self, *args, **kwargs):
|
||||
super(Pool, self).__init__(*args, **kwargs)
|
||||
|
||||
# TODO(comstud): Timeout connections not used in a while
|
||||
def create(self):
|
||||
LOG.debug('Pool creating new connection')
|
||||
return ConnectionClass()
|
||||
|
||||
|
||||
class ConnectionContext(rpc_common.Connection):
|
||||
"""The class that is actually returned to the caller of
|
||||
create_connection(). This is a essentially a wrapper around
|
||||
Connection that supports 'with' and can return a new Connection or
|
||||
one from a pool. It will also catch when an instance of this class
|
||||
is to be deleted so that we can return Connections to the pool on
|
||||
exceptions and so forth without making the caller be responsible for
|
||||
catching all exceptions and making sure to return a connection to
|
||||
the pool.
|
||||
"""
|
||||
|
||||
_connection_pool = Pool(max_size=FLAGS.rpc_conn_pool_size,
|
||||
order_as_stack=True)
|
||||
|
||||
def __init__(self, pooled=True):
|
||||
"""Create a new connection, or get one from the pool"""
|
||||
self.connection = None
|
||||
if pooled:
|
||||
self.connection = self._connection_pool.get()
|
||||
else:
|
||||
self.connection = ConnectionClass()
|
||||
self.pooled = pooled
|
||||
|
||||
def __enter__(self):
|
||||
"""When with ConnectionContext() is used, return self"""
|
||||
return self
|
||||
|
||||
def _done(self):
|
||||
"""If the connection came from a pool, clean it up and put it back.
|
||||
If it did not come from a pool, close it.
|
||||
"""
|
||||
if self.connection:
|
||||
if self.pooled:
|
||||
# Reset the connection so it's ready for the next caller
|
||||
# to grab from the pool
|
||||
self.connection.reset()
|
||||
self._connection_pool.put(self.connection)
|
||||
else:
|
||||
try:
|
||||
self.connection.close()
|
||||
except Exception:
|
||||
pass
|
||||
self.connection = None
|
||||
|
||||
def __exit__(self, exc_type, exc_value, tb):
|
||||
"""End of 'with' statement. We're done here."""
|
||||
self._done()
|
||||
|
||||
def __del__(self):
|
||||
"""Caller is done with this connection. Make sure we cleaned up."""
|
||||
self._done()
|
||||
|
||||
def close(self):
|
||||
"""Caller is done with this connection."""
|
||||
self._done()
|
||||
|
||||
def create_consumer(self, topic, proxy, fanout=False):
|
||||
self.connection.create_consumer(topic, proxy, fanout)
|
||||
|
||||
def consume_in_thread(self):
|
||||
self.connection.consume_in_thread()
|
||||
|
||||
def __getattr__(self, key):
|
||||
"""Proxy all other calls to the Connection instance"""
|
||||
if self.connection:
|
||||
return getattr(self.connection, key)
|
||||
else:
|
||||
raise exception.InvalidRPCConnectionReuse()
|
||||
|
||||
|
||||
def msg_reply(msg_id, reply=None, failure=None, ending=False):
|
||||
"""Sends a reply or an error on the channel signified by msg_id.
|
||||
|
||||
Failure should be a sys.exc_info() tuple.
|
||||
|
||||
"""
|
||||
with ConnectionContext() as conn:
|
||||
if failure:
|
||||
message = str(failure[1])
|
||||
tb = traceback.format_exception(*failure)
|
||||
LOG.error(_("Returning exception %s to caller"), message)
|
||||
LOG.error(tb)
|
||||
failure = (failure[0].__name__, str(failure[1]), tb)
|
||||
|
||||
try:
|
||||
msg = {'result': reply, 'failure': failure}
|
||||
except TypeError:
|
||||
msg = {'result': dict((k, repr(v))
|
||||
for k, v in reply.__dict__.iteritems()),
|
||||
'failure': failure}
|
||||
if ending:
|
||||
msg['ending'] = True
|
||||
conn.direct_send(msg_id, msg)
|
||||
|
||||
|
||||
class RpcContext(context.RequestContext):
|
||||
"""Context that supports replying to a rpc.call"""
|
||||
def __init__(self, *args, **kwargs):
|
||||
msg_id = kwargs.pop('msg_id', None)
|
||||
self.msg_id = msg_id
|
||||
super(RpcContext, self).__init__(*args, **kwargs)
|
||||
|
||||
def reply(self, reply=None, failure=None, ending=False):
|
||||
if self.msg_id:
|
||||
msg_reply(self.msg_id, reply, failure, ending)
|
||||
if ending:
|
||||
self.msg_id = None
|
||||
|
||||
|
||||
def unpack_context(msg):
|
||||
"""Unpack context from msg."""
|
||||
context_dict = {}
|
||||
for key in list(msg.keys()):
|
||||
# NOTE(vish): Some versions of python don't like unicode keys
|
||||
# in kwargs.
|
||||
key = str(key)
|
||||
if key.startswith('_context_'):
|
||||
value = msg.pop(key)
|
||||
context_dict[key[9:]] = value
|
||||
context_dict['msg_id'] = msg.pop('_msg_id', None)
|
||||
ctx = RpcContext.from_dict(context_dict)
|
||||
LOG.debug(_('unpacked context: %s'), ctx.to_dict())
|
||||
return ctx
|
||||
|
||||
|
||||
def pack_context(msg, context):
|
||||
"""Pack context into msg.
|
||||
|
||||
Values for message keys need to be less than 255 chars, so we pull
|
||||
context out into a bunch of separate keys. If we want to support
|
||||
more arguments in rabbit messages, we may want to do the same
|
||||
for args at some point.
|
||||
|
||||
"""
|
||||
context_d = dict([('_context_%s' % key, value)
|
||||
for (key, value) in context.to_dict().iteritems()])
|
||||
msg.update(context_d)
|
||||
|
||||
|
||||
class ProxyCallback(object):
|
||||
"""Calls methods on a proxy object based on method and args."""
|
||||
|
||||
def __init__(self, proxy):
|
||||
self.proxy = proxy
|
||||
self.pool = greenpool.GreenPool(FLAGS.rpc_thread_pool_size)
|
||||
|
||||
def __call__(self, message_data):
|
||||
"""Consumer callback to call a method on a proxy object.
|
||||
|
||||
Parses the message for validity and fires off a thread to call the
|
||||
proxy object method.
|
||||
|
||||
Message data should be a dictionary with two keys:
|
||||
method: string representing the method to call
|
||||
args: dictionary of arg: value
|
||||
|
||||
Example: {'method': 'echo', 'args': {'value': 42}}
|
||||
|
||||
"""
|
||||
# It is important to clear the context here, because at this point
|
||||
# the previous context is stored in local.store.context
|
||||
if hasattr(local.store, 'context'):
|
||||
del local.store.context
|
||||
rpc_common._safe_log(LOG.debug, _('received %s'), message_data)
|
||||
ctxt = unpack_context(message_data)
|
||||
method = message_data.get('method')
|
||||
args = message_data.get('args', {})
|
||||
if not method:
|
||||
LOG.warn(_('no method for message: %s') % message_data)
|
||||
ctxt.reply(_('No method for message: %s') % message_data)
|
||||
return
|
||||
self.pool.spawn_n(self._process_data, ctxt, method, args)
|
||||
|
||||
@exception.wrap_exception()
|
||||
def _process_data(self, ctxt, method, args):
|
||||
"""Thread that magically looks for a method on the proxy
|
||||
object and calls it.
|
||||
"""
|
||||
|
||||
try:
|
||||
node_func = getattr(self.proxy, str(method))
|
||||
node_args = dict((str(k), v) for k, v in args.iteritems())
|
||||
# NOTE(vish): magic is fun!
|
||||
rval = node_func(context=ctxt, **node_args)
|
||||
# Check if the result was a generator
|
||||
if inspect.isgenerator(rval):
|
||||
for x in rval:
|
||||
ctxt.reply(x, None)
|
||||
else:
|
||||
ctxt.reply(rval, None)
|
||||
# This final None tells multicall that it is done.
|
||||
ctxt.reply(ending=True)
|
||||
except Exception as e:
|
||||
LOG.exception('Exception during message handling')
|
||||
ctxt.reply(None, sys.exc_info())
|
||||
return
|
||||
|
||||
|
||||
class MulticallWaiter(object):
|
||||
def __init__(self, connection):
|
||||
self._connection = connection
|
||||
self._iterator = connection.iterconsume()
|
||||
self._result = None
|
||||
self._done = False
|
||||
self._got_ending = False
|
||||
|
||||
def done(self):
|
||||
if self._done:
|
||||
return
|
||||
self._done = True
|
||||
self._iterator.close()
|
||||
self._iterator = None
|
||||
self._connection.close()
|
||||
|
||||
def __call__(self, data):
|
||||
"""The consume() callback will call this. Store the result."""
|
||||
if data['failure']:
|
||||
self._result = rpc_common.RemoteError(*data['failure'])
|
||||
elif data.get('ending', False):
|
||||
self._got_ending = True
|
||||
else:
|
||||
self._result = data['result']
|
||||
|
||||
def __iter__(self):
|
||||
"""Return a result until we get a 'None' response from consumer"""
|
||||
if self._done:
|
||||
raise StopIteration
|
||||
while True:
|
||||
self._iterator.next()
|
||||
if self._got_ending:
|
||||
self.done()
|
||||
raise StopIteration
|
||||
result = self._result
|
||||
if isinstance(result, Exception):
|
||||
self.done()
|
||||
raise result
|
||||
yield result
|
||||
|
||||
|
||||
def create_connection(new=True):
|
||||
"""Create a connection"""
|
||||
return ConnectionContext(pooled=not new)
|
||||
|
||||
|
||||
def multicall(context, topic, msg):
|
||||
"""Make a call that returns multiple times."""
|
||||
# Can't use 'with' for multicall, as it returns an iterator
|
||||
# that will continue to use the connection. When it's done,
|
||||
# connection.close() will get called which will put it back into
|
||||
# the pool
|
||||
LOG.debug(_('Making asynchronous call on %s ...'), topic)
|
||||
msg_id = uuid.uuid4().hex
|
||||
msg.update({'_msg_id': msg_id})
|
||||
LOG.debug(_('MSG_ID is %s') % (msg_id))
|
||||
pack_context(msg, context)
|
||||
|
||||
conn = ConnectionContext()
|
||||
wait_msg = MulticallWaiter(conn)
|
||||
conn.declare_direct_consumer(msg_id, wait_msg)
|
||||
conn.topic_send(topic, msg)
|
||||
return wait_msg
|
||||
|
||||
|
||||
def call(context, topic, msg):
|
||||
"""Sends a message on a topic and wait for a response."""
|
||||
rv = multicall(context, topic, msg)
|
||||
# NOTE(vish): return the last result from the multicall
|
||||
rv = list(rv)
|
||||
if not rv:
|
||||
return
|
||||
return rv[-1]
|
||||
|
||||
|
||||
def cast(context, topic, msg):
|
||||
"""Sends a message on a topic without waiting for a response."""
|
||||
LOG.debug(_('Making asynchronous cast on %s...'), topic)
|
||||
pack_context(msg, context)
|
||||
with ConnectionContext() as conn:
|
||||
conn.topic_send(topic, msg)
|
||||
|
||||
|
||||
def fanout_cast(context, topic, msg):
|
||||
"""Sends a message on a fanout exchange without waiting for a response."""
|
||||
LOG.debug(_('Making asynchronous fanout cast...'))
|
||||
pack_context(msg, context)
|
||||
with ConnectionContext() as conn:
|
||||
conn.fanout_send(topic, msg)
|
||||
|
||||
|
||||
def notify(context, topic, msg):
|
||||
"""Sends a notification event on a topic."""
|
||||
LOG.debug(_('Sending notification on %s...'), topic)
|
||||
pack_context(msg, context)
|
||||
with ConnectionContext() as conn:
|
||||
conn.notify_send(topic, msg, durable=True)
|
||||
@@ -23,6 +23,7 @@ from nova import exception
|
||||
from nova import flags
|
||||
from nova import log as logging
|
||||
|
||||
|
||||
LOG = logging.getLogger('nova.rpc')
|
||||
|
||||
flags.DEFINE_integer('rpc_thread_pool_size', 1024,
|
||||
|
||||
@@ -14,16 +14,12 @@
|
||||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
import inspect
|
||||
import itertools
|
||||
import sys
|
||||
import time
|
||||
import traceback
|
||||
import uuid
|
||||
|
||||
import eventlet
|
||||
from eventlet import greenpool
|
||||
from eventlet import pools
|
||||
import greenlet
|
||||
import kombu
|
||||
import kombu.entity
|
||||
@@ -33,8 +29,9 @@ import kombu.connection
|
||||
from nova import context
|
||||
from nova import exception
|
||||
from nova import flags
|
||||
from nova import local
|
||||
from nova.rpc import common as rpc_common
|
||||
from nova.rpc import amqp as rpc_amqp
|
||||
|
||||
|
||||
FLAGS = flags.FLAGS
|
||||
LOG = rpc_common.LOG
|
||||
@@ -584,320 +581,40 @@ class Connection(object):
|
||||
def create_consumer(self, topic, proxy, fanout=False):
|
||||
"""Create a consumer that calls a method in a proxy object"""
|
||||
if fanout:
|
||||
self.declare_fanout_consumer(topic, ProxyCallback(proxy))
|
||||
self.declare_fanout_consumer(topic,
|
||||
rpc_amqp.ProxyCallback(proxy))
|
||||
else:
|
||||
self.declare_topic_consumer(topic, ProxyCallback(proxy))
|
||||
self.declare_topic_consumer(topic, rpc_amqp.ProxyCallback(proxy))
|
||||
|
||||
|
||||
class Pool(pools.Pool):
|
||||
"""Class that implements a Pool of Connections."""
|
||||
|
||||
# TODO(comstud): Timeout connections not used in a while
|
||||
def create(self):
|
||||
LOG.debug('Pool creating new connection')
|
||||
return Connection()
|
||||
|
||||
# Create a ConnectionPool to use for RPC calls. We'll order the
|
||||
# pool as a stack (LIFO), so that we can potentially loop through and
|
||||
# timeout old unused connections at some point
|
||||
ConnectionPool = Pool(
|
||||
max_size=FLAGS.rpc_conn_pool_size,
|
||||
order_as_stack=True)
|
||||
|
||||
|
||||
class ConnectionContext(rpc_common.Connection):
|
||||
"""The class that is actually returned to the caller of
|
||||
create_connection(). This is a essentially a wrapper around
|
||||
Connection that supports 'with' and can return a new Connection or
|
||||
one from a pool. It will also catch when an instance of this class
|
||||
is to be deleted so that we can return Connections to the pool on
|
||||
exceptions and so forth without making the caller be responsible for
|
||||
catching all exceptions and making sure to return a connection to
|
||||
the pool.
|
||||
"""
|
||||
|
||||
def __init__(self, pooled=True):
|
||||
"""Create a new connection, or get one from the pool"""
|
||||
self.connection = None
|
||||
if pooled:
|
||||
self.connection = ConnectionPool.get()
|
||||
else:
|
||||
self.connection = Connection()
|
||||
self.pooled = pooled
|
||||
|
||||
def __enter__(self):
|
||||
"""with ConnectionContext() should return self"""
|
||||
return self
|
||||
|
||||
def _done(self):
|
||||
"""If the connection came from a pool, clean it up and put it back.
|
||||
If it did not come from a pool, close it.
|
||||
"""
|
||||
if self.connection:
|
||||
if self.pooled:
|
||||
# Reset the connection so it's ready for the next caller
|
||||
# to grab from the pool
|
||||
self.connection.reset()
|
||||
ConnectionPool.put(self.connection)
|
||||
else:
|
||||
try:
|
||||
self.connection.close()
|
||||
except Exception:
|
||||
# There's apparently a bug in kombu 'memory' transport
|
||||
# which causes an assert failure.
|
||||
# But, we probably want to ignore all exceptions when
|
||||
# trying to close a connection, anyway...
|
||||
pass
|
||||
self.connection = None
|
||||
|
||||
def __exit__(self, t, v, tb):
|
||||
"""end of 'with' statement. We're done here."""
|
||||
self._done()
|
||||
|
||||
def __del__(self):
|
||||
"""Caller is done with this connection. Make sure we cleaned up."""
|
||||
self._done()
|
||||
|
||||
def close(self):
|
||||
"""Caller is done with this connection."""
|
||||
self._done()
|
||||
|
||||
def create_consumer(self, topic, proxy, fanout=False):
|
||||
self.connection.create_consumer(topic, proxy, fanout)
|
||||
|
||||
def consume_in_thread(self):
|
||||
self.connection.consume_in_thread()
|
||||
|
||||
def __getattr__(self, key):
|
||||
"""Proxy all other calls to the Connection instance"""
|
||||
if self.connection:
|
||||
return getattr(self.connection, key)
|
||||
else:
|
||||
raise exception.InvalidRPCConnectionReuse()
|
||||
|
||||
|
||||
class ProxyCallback(object):
|
||||
"""Calls methods on a proxy object based on method and args."""
|
||||
|
||||
def __init__(self, proxy):
|
||||
self.proxy = proxy
|
||||
self.pool = greenpool.GreenPool(FLAGS.rpc_thread_pool_size)
|
||||
|
||||
def __call__(self, message_data):
|
||||
"""Consumer callback to call a method on a proxy object.
|
||||
|
||||
Parses the message for validity and fires off a thread to call the
|
||||
proxy object method.
|
||||
|
||||
Message data should be a dictionary with two keys:
|
||||
method: string representing the method to call
|
||||
args: dictionary of arg: value
|
||||
|
||||
Example: {'method': 'echo', 'args': {'value': 42}}
|
||||
|
||||
"""
|
||||
# It is important to clear the context here, because at this point
|
||||
# the previous context is stored in local.store.context
|
||||
if hasattr(local.store, 'context'):
|
||||
del local.store.context
|
||||
rpc_common._safe_log(LOG.debug, _('received %s'), message_data)
|
||||
ctxt = _unpack_context(message_data)
|
||||
method = message_data.get('method')
|
||||
args = message_data.get('args', {})
|
||||
if not method:
|
||||
LOG.warn(_('no method for message: %s') % message_data)
|
||||
ctxt.reply(_('No method for message: %s') % message_data)
|
||||
return
|
||||
self.pool.spawn_n(self._process_data, ctxt, method, args)
|
||||
|
||||
@exception.wrap_exception()
|
||||
def _process_data(self, ctxt, method, args):
|
||||
"""Thread that magically looks for a method on the proxy
|
||||
object and calls it.
|
||||
"""
|
||||
|
||||
try:
|
||||
node_func = getattr(self.proxy, str(method))
|
||||
node_args = dict((str(k), v) for k, v in args.iteritems())
|
||||
# NOTE(vish): magic is fun!
|
||||
rval = node_func(context=ctxt, **node_args)
|
||||
# Check if the result was a generator
|
||||
if inspect.isgenerator(rval):
|
||||
for x in rval:
|
||||
ctxt.reply(x, None)
|
||||
else:
|
||||
ctxt.reply(rval, None)
|
||||
# This final None tells multicall that it is done.
|
||||
ctxt.reply(ending=True)
|
||||
except Exception as e:
|
||||
LOG.exception('Exception during message handling')
|
||||
ctxt.reply(None, sys.exc_info())
|
||||
return
|
||||
|
||||
|
||||
def _unpack_context(msg):
|
||||
"""Unpack context from msg."""
|
||||
context_dict = {}
|
||||
for key in list(msg.keys()):
|
||||
# NOTE(vish): Some versions of python don't like unicode keys
|
||||
# in kwargs.
|
||||
key = str(key)
|
||||
if key.startswith('_context_'):
|
||||
value = msg.pop(key)
|
||||
context_dict[key[9:]] = value
|
||||
context_dict['msg_id'] = msg.pop('_msg_id', None)
|
||||
ctx = RpcContext.from_dict(context_dict)
|
||||
LOG.debug(_('unpacked context: %s'), ctx.to_dict())
|
||||
return ctx
|
||||
|
||||
|
||||
def _pack_context(msg, context):
|
||||
"""Pack context into msg.
|
||||
|
||||
Values for message keys need to be less than 255 chars, so we pull
|
||||
context out into a bunch of separate keys. If we want to support
|
||||
more arguments in rabbit messages, we may want to do the same
|
||||
for args at some point.
|
||||
|
||||
"""
|
||||
context_d = dict([('_context_%s' % key, value)
|
||||
for (key, value) in context.to_dict().iteritems()])
|
||||
msg.update(context_d)
|
||||
|
||||
|
||||
class RpcContext(context.RequestContext):
|
||||
"""Context that supports replying to a rpc.call"""
|
||||
def __init__(self, *args, **kwargs):
|
||||
msg_id = kwargs.pop('msg_id', None)
|
||||
self.msg_id = msg_id
|
||||
super(RpcContext, self).__init__(*args, **kwargs)
|
||||
|
||||
def reply(self, reply=None, failure=None, ending=False):
|
||||
if self.msg_id:
|
||||
msg_reply(self.msg_id, reply, failure, ending)
|
||||
if ending:
|
||||
self.msg_id = None
|
||||
|
||||
|
||||
class MulticallWaiter(object):
|
||||
def __init__(self, connection):
|
||||
self._connection = connection
|
||||
self._iterator = connection.iterconsume()
|
||||
self._result = None
|
||||
self._done = False
|
||||
self._got_ending = False
|
||||
|
||||
def done(self):
|
||||
if self._done:
|
||||
return
|
||||
self._done = True
|
||||
self._iterator.close()
|
||||
self._iterator = None
|
||||
self._connection.close()
|
||||
|
||||
def __call__(self, data):
|
||||
"""The consume() callback will call this. Store the result."""
|
||||
if data['failure']:
|
||||
self._result = rpc_common.RemoteError(*data['failure'])
|
||||
elif data.get('ending', False):
|
||||
self._got_ending = True
|
||||
else:
|
||||
self._result = data['result']
|
||||
|
||||
def __iter__(self):
|
||||
"""Return a result until we get a 'None' response from consumer"""
|
||||
if self._done:
|
||||
raise StopIteration
|
||||
while True:
|
||||
self._iterator.next()
|
||||
if self._got_ending:
|
||||
self.done()
|
||||
raise StopIteration
|
||||
result = self._result
|
||||
if isinstance(result, Exception):
|
||||
self.done()
|
||||
raise result
|
||||
yield result
|
||||
rpc_amqp.ConnectionClass = Connection
|
||||
|
||||
|
||||
def create_connection(new=True):
|
||||
"""Create a connection"""
|
||||
return ConnectionContext(pooled=not new)
|
||||
return rpc_amqp.create_connection(new)
|
||||
|
||||
|
||||
def multicall(context, topic, msg):
|
||||
"""Make a call that returns multiple times."""
|
||||
# Can't use 'with' for multicall, as it returns an iterator
|
||||
# that will continue to use the connection. When it's done,
|
||||
# connection.close() will get called which will put it back into
|
||||
# the pool
|
||||
LOG.debug(_('Making asynchronous call on %s ...'), topic)
|
||||
msg_id = uuid.uuid4().hex
|
||||
msg.update({'_msg_id': msg_id})
|
||||
LOG.debug(_('MSG_ID is %s') % (msg_id))
|
||||
_pack_context(msg, context)
|
||||
|
||||
conn = ConnectionContext()
|
||||
wait_msg = MulticallWaiter(conn)
|
||||
conn.declare_direct_consumer(msg_id, wait_msg)
|
||||
conn.topic_send(topic, msg)
|
||||
return wait_msg
|
||||
return rpc_amqp.multicall(context, topic, msg)
|
||||
|
||||
|
||||
def call(context, topic, msg):
|
||||
"""Sends a message on a topic and wait for a response."""
|
||||
rv = multicall(context, topic, msg)
|
||||
# NOTE(vish): return the last result from the multicall
|
||||
rv = list(rv)
|
||||
if not rv:
|
||||
return
|
||||
return rv[-1]
|
||||
return rpc_amqp.call(context, topic, msg)
|
||||
|
||||
|
||||
def cast(context, topic, msg):
|
||||
"""Sends a message on a topic without waiting for a response."""
|
||||
LOG.debug(_('Making asynchronous cast on %s...'), topic)
|
||||
_pack_context(msg, context)
|
||||
with ConnectionContext() as conn:
|
||||
conn.topic_send(topic, msg)
|
||||
return rpc_amqp.cast(context, topic, msg)
|
||||
|
||||
|
||||
def fanout_cast(context, topic, msg):
|
||||
"""Sends a message on a fanout exchange without waiting for a response."""
|
||||
LOG.debug(_('Making asynchronous fanout cast...'))
|
||||
_pack_context(msg, context)
|
||||
with ConnectionContext() as conn:
|
||||
conn.fanout_send(topic, msg)
|
||||
return rpc_amqp.fanout_cast(context, topic, msg)
|
||||
|
||||
|
||||
def notify(context, topic, msg):
|
||||
"""Sends a notification event on a topic."""
|
||||
LOG.debug(_('Sending notification on %s...'), topic)
|
||||
_pack_context(msg, context)
|
||||
with ConnectionContext() as conn:
|
||||
conn.notify_send(topic, msg, durable=True)
|
||||
|
||||
|
||||
def msg_reply(msg_id, reply=None, failure=None, ending=False):
|
||||
"""Sends a reply or an error on the channel signified by msg_id.
|
||||
|
||||
Failure should be a sys.exc_info() tuple.
|
||||
|
||||
"""
|
||||
with ConnectionContext() as conn:
|
||||
if failure:
|
||||
message = str(failure[1])
|
||||
tb = traceback.format_exception(*failure)
|
||||
LOG.error(_("Returning exception %s to caller"), message)
|
||||
LOG.error(tb)
|
||||
failure = (failure[0].__name__, str(failure[1]), tb)
|
||||
|
||||
try:
|
||||
msg = {'result': reply, 'failure': failure}
|
||||
except TypeError:
|
||||
msg = {'result': dict((k, repr(v))
|
||||
for k, v in reply.__dict__.iteritems()),
|
||||
'failure': failure}
|
||||
if ending:
|
||||
msg['ending'] = True
|
||||
conn.direct_send(msg_id, msg)
|
||||
return rpc_amqp.notify(context, topic, msg)
|
||||
|
||||
485
nova/rpc/impl_qpid.py
Normal file
485
nova/rpc/impl_qpid.py
Normal file
@@ -0,0 +1,485 @@
|
||||
# vim: tabstop=4 shiftwidth=4 softtabstop=4
|
||||
|
||||
# Copyright 2011 OpenStack LLC
|
||||
# Copyright 2011 - 2012, 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
|
||||
# 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.
|
||||
|
||||
import itertools
|
||||
import time
|
||||
import uuid
|
||||
import json
|
||||
|
||||
import eventlet
|
||||
import greenlet
|
||||
import qpid.messaging
|
||||
import qpid.messaging.exceptions
|
||||
|
||||
from nova import flags
|
||||
from nova.rpc import amqp as rpc_amqp
|
||||
from nova.rpc.common import LOG
|
||||
|
||||
|
||||
flags.DEFINE_string('qpid_hostname', 'localhost', 'Qpid broker hostname')
|
||||
flags.DEFINE_string('qpid_port', '5672', 'Qpid broker port')
|
||||
flags.DEFINE_string('qpid_username', '', 'Username for qpid connection')
|
||||
flags.DEFINE_string('qpid_password', '', 'Password for qpid connection')
|
||||
flags.DEFINE_string('qpid_sasl_mechanisms', '',
|
||||
'Space separated list of SASL mechanisms to use for auth')
|
||||
flags.DEFINE_boolean('qpid_reconnect', True, 'Automatically reconnect')
|
||||
flags.DEFINE_integer('qpid_reconnect_timeout', 0,
|
||||
'Reconnection timeout in seconds')
|
||||
flags.DEFINE_integer('qpid_reconnect_limit', 0,
|
||||
'Max reconnections before giving up')
|
||||
flags.DEFINE_integer('qpid_reconnect_interval_min', 0,
|
||||
'Minimum seconds between reconnection attempts')
|
||||
flags.DEFINE_integer('qpid_reconnect_interval_max', 0,
|
||||
'Maximum seconds between reconnection attempts')
|
||||
flags.DEFINE_integer('qpid_reconnect_interval', 0,
|
||||
'Equivalent to setting max and min to the same value')
|
||||
flags.DEFINE_integer('qpid_heartbeat', 5,
|
||||
'Seconds between heartbeats used to keep the connection alive')
|
||||
flags.DEFINE_string('qpid_protocol', 'tcp',
|
||||
"Transport to use, either 'tcp' or 'ssl'")
|
||||
flags.DEFINE_boolean('qpid_tcp_nodelay', True, 'Disable Nagle algorithm')
|
||||
|
||||
|
||||
FLAGS = flags.FLAGS
|
||||
|
||||
|
||||
class ConsumerBase(object):
|
||||
"""Consumer base class."""
|
||||
|
||||
def __init__(self, session, callback, node_name, node_opts,
|
||||
link_name, link_opts):
|
||||
"""Declare a queue on an amqp session.
|
||||
|
||||
'session' is the amqp session to use
|
||||
'callback' is the callback to call when messages are received
|
||||
'node_name' is the first part of the Qpid address string, before ';'
|
||||
'node_opts' will be applied to the "x-declare" section of "node"
|
||||
in the address string.
|
||||
'link_name' goes into the "name" field of the "link" in the address
|
||||
string
|
||||
'link_opts' will be applied to the "x-declare" section of "link"
|
||||
in the address string.
|
||||
"""
|
||||
self.callback = callback
|
||||
self.receiver = None
|
||||
self.session = None
|
||||
|
||||
addr_opts = {
|
||||
"create": "always",
|
||||
"node": {
|
||||
"type": "topic",
|
||||
"x-declare": {
|
||||
"durable": True,
|
||||
"auto-delete": True,
|
||||
},
|
||||
},
|
||||
"link": {
|
||||
"name": link_name,
|
||||
"durable": True,
|
||||
"x-declare": {
|
||||
"durable": False,
|
||||
"auto-delete": True,
|
||||
"exclusive": False,
|
||||
},
|
||||
},
|
||||
}
|
||||
addr_opts["node"]["x-declare"].update(node_opts)
|
||||
addr_opts["link"]["x-declare"].update(link_opts)
|
||||
|
||||
self.address = "%s ; %s" % (node_name, json.dumps(addr_opts))
|
||||
|
||||
self.reconnect(session)
|
||||
|
||||
def reconnect(self, session):
|
||||
"""Re-declare the receiver after a qpid reconnect"""
|
||||
self.session = session
|
||||
self.receiver = session.receiver(self.address)
|
||||
self.receiver.capacity = 1
|
||||
|
||||
def consume(self):
|
||||
"""Fetch the message and pass it to the callback object"""
|
||||
message = self.receiver.fetch()
|
||||
self.callback(message.content)
|
||||
|
||||
def get_receiver(self):
|
||||
return self.receiver
|
||||
|
||||
|
||||
class DirectConsumer(ConsumerBase):
|
||||
"""Queue/consumer class for 'direct'"""
|
||||
|
||||
def __init__(self, session, msg_id, callback):
|
||||
"""Init a 'direct' queue.
|
||||
|
||||
'session' is the amqp session to use
|
||||
'msg_id' is the msg_id to listen on
|
||||
'callback' is the callback to call when messages are received
|
||||
"""
|
||||
|
||||
super(DirectConsumer, self).__init__(session, callback,
|
||||
"%s/%s" % (msg_id, msg_id),
|
||||
{"type": "direct"},
|
||||
msg_id,
|
||||
{"exclusive": True})
|
||||
|
||||
|
||||
class TopicConsumer(ConsumerBase):
|
||||
"""Consumer class for 'topic'"""
|
||||
|
||||
def __init__(self, session, topic, callback):
|
||||
"""Init a 'topic' queue.
|
||||
|
||||
'session' is the amqp session to use
|
||||
'topic' is the topic to listen on
|
||||
'callback' is the callback to call when messages are received
|
||||
"""
|
||||
|
||||
super(TopicConsumer, self).__init__(session, callback,
|
||||
"%s/%s" % (FLAGS.control_exchange, topic), {},
|
||||
topic, {})
|
||||
|
||||
|
||||
class FanoutConsumer(ConsumerBase):
|
||||
"""Consumer class for 'fanout'"""
|
||||
|
||||
def __init__(self, session, topic, callback):
|
||||
"""Init a 'fanout' queue.
|
||||
|
||||
'session' is the amqp session to use
|
||||
'topic' is the topic to listen on
|
||||
'callback' is the callback to call when messages are received
|
||||
"""
|
||||
|
||||
super(FanoutConsumer, self).__init__(session, callback,
|
||||
"%s_fanout" % topic,
|
||||
{"durable": False, "type": "fanout"},
|
||||
"%s_fanout_%s" % (topic, uuid.uuid4().hex),
|
||||
{"exclusive": True})
|
||||
|
||||
|
||||
class Publisher(object):
|
||||
"""Base Publisher class"""
|
||||
|
||||
def __init__(self, session, node_name, node_opts=None):
|
||||
"""Init the Publisher class with the exchange_name, routing_key,
|
||||
and other options
|
||||
"""
|
||||
self.sender = None
|
||||
self.session = session
|
||||
|
||||
addr_opts = {
|
||||
"create": "always",
|
||||
"node": {
|
||||
"type": "topic",
|
||||
"x-declare": {
|
||||
"durable": False,
|
||||
# auto-delete isn't implemented for exchanges in qpid,
|
||||
# but put in here anyway
|
||||
"auto-delete": True,
|
||||
},
|
||||
},
|
||||
}
|
||||
if node_opts:
|
||||
addr_opts["node"]["x-declare"].update(node_opts)
|
||||
|
||||
self.address = "%s ; %s" % (node_name, json.dumps(addr_opts))
|
||||
|
||||
self.reconnect(session)
|
||||
|
||||
def reconnect(self, session):
|
||||
"""Re-establish the Sender after a reconnection"""
|
||||
self.sender = session.sender(self.address)
|
||||
|
||||
def send(self, msg):
|
||||
"""Send a message"""
|
||||
self.sender.send(msg)
|
||||
|
||||
|
||||
class DirectPublisher(Publisher):
|
||||
"""Publisher class for 'direct'"""
|
||||
def __init__(self, session, msg_id):
|
||||
"""Init a 'direct' publisher."""
|
||||
super(DirectPublisher, self).__init__(session, msg_id,
|
||||
{"type": "Direct"})
|
||||
|
||||
|
||||
class TopicPublisher(Publisher):
|
||||
"""Publisher class for 'topic'"""
|
||||
def __init__(self, session, topic):
|
||||
"""init a 'topic' publisher.
|
||||
"""
|
||||
super(TopicPublisher, self).__init__(session,
|
||||
"%s/%s" % (FLAGS.control_exchange, topic))
|
||||
|
||||
|
||||
class FanoutPublisher(Publisher):
|
||||
"""Publisher class for 'fanout'"""
|
||||
def __init__(self, session, topic):
|
||||
"""init a 'fanout' publisher.
|
||||
"""
|
||||
super(FanoutPublisher, self).__init__(session,
|
||||
"%s_fanout" % topic, {"type": "fanout"})
|
||||
|
||||
|
||||
class NotifyPublisher(Publisher):
|
||||
"""Publisher class for notifications"""
|
||||
def __init__(self, session, topic):
|
||||
"""init a 'topic' publisher.
|
||||
"""
|
||||
super(NotifyPublisher, self).__init__(session,
|
||||
"%s/%s" % (FLAGS.control_exchange, topic),
|
||||
{"durable": True})
|
||||
|
||||
|
||||
class Connection(object):
|
||||
"""Connection object."""
|
||||
|
||||
def __init__(self):
|
||||
self.session = None
|
||||
self.consumers = {}
|
||||
self.consumer_thread = None
|
||||
|
||||
self.broker = FLAGS.qpid_hostname + ":" + FLAGS.qpid_port
|
||||
# Create the connection - this does not open the connection
|
||||
self.connection = qpid.messaging.Connection(self.broker)
|
||||
|
||||
# Check if flags are set and if so set them for the connection
|
||||
# before we call open
|
||||
self.connection.username = FLAGS.qpid_username
|
||||
self.connection.password = FLAGS.qpid_password
|
||||
self.connection.sasl_mechanisms = FLAGS.qpid_sasl_mechanisms
|
||||
self.connection.reconnect = FLAGS.qpid_reconnect
|
||||
self.connection.reconnect_timeout = FLAGS.qpid_reconnect_timeout
|
||||
self.connection.reconnect_limit = FLAGS.qpid_reconnect_limit
|
||||
self.connection.reconnect_interval_max = \
|
||||
FLAGS.qpid_reconnect_interval_max
|
||||
self.connection.reconnect_interval_min = \
|
||||
FLAGS.qpid_reconnect_interval_min
|
||||
self.connection.reconnect_interval = FLAGS.qpid_reconnect_interval
|
||||
self.connection.hearbeat = FLAGS.qpid_heartbeat
|
||||
self.connection.protocol = FLAGS.qpid_protocol
|
||||
self.connection.tcp_nodelay = FLAGS.qpid_tcp_nodelay
|
||||
|
||||
# Open is part of reconnect -
|
||||
# NOTE(WGH) not sure we need this with the reconnect flags
|
||||
self.reconnect()
|
||||
|
||||
def _register_consumer(self, consumer):
|
||||
self.consumers[str(consumer.get_receiver())] = consumer
|
||||
|
||||
def _lookup_consumer(self, receiver):
|
||||
return self.consumers[str(receiver)]
|
||||
|
||||
def reconnect(self):
|
||||
"""Handles reconnecting and re-establishing sessions and queues"""
|
||||
if self.connection.opened():
|
||||
try:
|
||||
self.connection.close()
|
||||
except qpid.messaging.exceptions.ConnectionError:
|
||||
pass
|
||||
|
||||
while True:
|
||||
try:
|
||||
self.connection.open()
|
||||
except qpid.messaging.exceptions.ConnectionError, e:
|
||||
LOG.error(_('Unable to connect to AMQP server: %s ' % str(e)))
|
||||
time.sleep(FLAGS.qpid_reconnect_interval or 1)
|
||||
else:
|
||||
break
|
||||
|
||||
LOG.info(_('Connected to AMQP server on %s' % self.broker))
|
||||
|
||||
self.session = self.connection.session()
|
||||
|
||||
for consumer in self.consumers.itervalues():
|
||||
consumer.reconnect(self.session)
|
||||
|
||||
if self.consumers:
|
||||
LOG.debug(_("Re-established AMQP queues"))
|
||||
|
||||
def ensure(self, error_callback, method, *args, **kwargs):
|
||||
while True:
|
||||
try:
|
||||
return method(*args, **kwargs)
|
||||
except qpid.messaging.exceptions.ConnectionError, e:
|
||||
if error_callback:
|
||||
error_callback(e)
|
||||
self.reconnect()
|
||||
|
||||
def close(self):
|
||||
"""Close/release this connection"""
|
||||
self.cancel_consumer_thread()
|
||||
self.connection.close()
|
||||
self.connection = None
|
||||
|
||||
def reset(self):
|
||||
"""Reset a connection so it can be used again"""
|
||||
self.cancel_consumer_thread()
|
||||
self.session.close()
|
||||
self.session = self.connection.session()
|
||||
self.consumers = {}
|
||||
|
||||
def declare_consumer(self, consumer_cls, topic, callback):
|
||||
"""Create a Consumer using the class that was passed in and
|
||||
add it to our list of consumers
|
||||
"""
|
||||
def _connect_error(exc):
|
||||
log_info = {'topic': topic, 'err_str': str(exc)}
|
||||
LOG.error(_("Failed to declare consumer for topic '%(topic)s': "
|
||||
"%(err_str)s") % log_info)
|
||||
|
||||
def _declare_consumer():
|
||||
consumer = consumer_cls(self.session, topic, callback)
|
||||
self._register_consumer(consumer)
|
||||
return consumer
|
||||
|
||||
return self.ensure(_connect_error, _declare_consumer)
|
||||
|
||||
def iterconsume(self, limit=None):
|
||||
"""Return an iterator that will consume from all queues/consumers"""
|
||||
|
||||
def _error_callback(exc):
|
||||
LOG.exception(_('Failed to consume message from queue: %s') %
|
||||
str(exc))
|
||||
|
||||
def _consume():
|
||||
nxt_receiver = self.session.next_receiver()
|
||||
self._lookup_consumer(nxt_receiver).consume()
|
||||
|
||||
for iteration in itertools.count(0):
|
||||
if limit and iteration >= limit:
|
||||
raise StopIteration
|
||||
yield self.ensure(_error_callback, _consume)
|
||||
|
||||
def cancel_consumer_thread(self):
|
||||
"""Cancel a consumer thread"""
|
||||
if self.consumer_thread is not None:
|
||||
self.consumer_thread.kill()
|
||||
try:
|
||||
self.consumer_thread.wait()
|
||||
except greenlet.GreenletExit:
|
||||
pass
|
||||
self.consumer_thread = None
|
||||
|
||||
def publisher_send(self, cls, topic, msg):
|
||||
"""Send to a publisher based on the publisher class"""
|
||||
|
||||
def _connect_error(exc):
|
||||
log_info = {'topic': topic, 'err_str': str(exc)}
|
||||
LOG.exception(_("Failed to publish message to topic "
|
||||
"'%(topic)s': %(err_str)s") % log_info)
|
||||
|
||||
def _publisher_send():
|
||||
publisher = cls(self.session, topic)
|
||||
publisher.send(msg)
|
||||
|
||||
return self.ensure(_connect_error, _publisher_send)
|
||||
|
||||
def declare_direct_consumer(self, topic, callback):
|
||||
"""Create a 'direct' queue.
|
||||
In nova's use, this is generally a msg_id queue used for
|
||||
responses for call/multicall
|
||||
"""
|
||||
self.declare_consumer(DirectConsumer, topic, callback)
|
||||
|
||||
def declare_topic_consumer(self, topic, callback=None):
|
||||
"""Create a 'topic' consumer."""
|
||||
self.declare_consumer(TopicConsumer, topic, callback)
|
||||
|
||||
def declare_fanout_consumer(self, topic, callback):
|
||||
"""Create a 'fanout' consumer"""
|
||||
self.declare_consumer(FanoutConsumer, topic, callback)
|
||||
|
||||
def direct_send(self, msg_id, msg):
|
||||
"""Send a 'direct' message"""
|
||||
self.publisher_send(DirectPublisher, msg_id, msg)
|
||||
|
||||
def topic_send(self, topic, msg):
|
||||
"""Send a 'topic' message"""
|
||||
self.publisher_send(TopicPublisher, topic, msg)
|
||||
|
||||
def fanout_send(self, topic, msg):
|
||||
"""Send a 'fanout' message"""
|
||||
self.publisher_send(FanoutPublisher, topic, msg)
|
||||
|
||||
def notify_send(self, topic, msg, **kwargs):
|
||||
"""Send a notify message on a topic"""
|
||||
self.publisher_send(NotifyPublisher, topic, msg)
|
||||
|
||||
def consume(self, limit=None):
|
||||
"""Consume from all queues/consumers"""
|
||||
it = self.iterconsume(limit=limit)
|
||||
while True:
|
||||
try:
|
||||
it.next()
|
||||
except StopIteration:
|
||||
return
|
||||
|
||||
def consume_in_thread(self):
|
||||
"""Consumer from all queues/consumers in a greenthread"""
|
||||
def _consumer_thread():
|
||||
try:
|
||||
self.consume()
|
||||
except greenlet.GreenletExit:
|
||||
return
|
||||
if self.consumer_thread is None:
|
||||
self.consumer_thread = eventlet.spawn(_consumer_thread)
|
||||
return self.consumer_thread
|
||||
|
||||
def create_consumer(self, topic, proxy, fanout=False):
|
||||
"""Create a consumer that calls a method in a proxy object"""
|
||||
if fanout:
|
||||
consumer = FanoutConsumer(self.session, topic,
|
||||
rpc_amqp.ProxyCallback(proxy))
|
||||
else:
|
||||
consumer = TopicConsumer(self.session, topic,
|
||||
rpc_amqp.ProxyCallback(proxy))
|
||||
self._register_consumer(consumer)
|
||||
return consumer
|
||||
|
||||
|
||||
rpc_amqp.ConnectionClass = Connection
|
||||
|
||||
|
||||
def create_connection(new=True):
|
||||
"""Create a connection"""
|
||||
return rpc_amqp.create_connection(new)
|
||||
|
||||
|
||||
def multicall(context, topic, msg):
|
||||
"""Make a call that returns multiple times."""
|
||||
return rpc_amqp.multicall(context, topic, msg)
|
||||
|
||||
|
||||
def call(context, topic, msg):
|
||||
"""Sends a message on a topic and wait for a response."""
|
||||
return rpc_amqp.call(context, topic, msg)
|
||||
|
||||
|
||||
def cast(context, topic, msg):
|
||||
"""Sends a message on a topic without waiting for a response."""
|
||||
return rpc_amqp.cast(context, topic, msg)
|
||||
|
||||
|
||||
def fanout_cast(context, topic, msg):
|
||||
"""Sends a message on a fanout exchange without waiting for a response."""
|
||||
return rpc_amqp.fanout_cast(context, topic, msg)
|
||||
|
||||
|
||||
def notify(context, topic, msg):
|
||||
"""Sends a notification event on a topic."""
|
||||
return rpc_amqp.notify(context, topic, msg)
|
||||
293
nova/tests/rpc/test_qpid.py
Normal file
293
nova/tests/rpc/test_qpid.py
Normal file
@@ -0,0 +1,293 @@
|
||||
# vim: tabstop=4 shiftwidth=4 softtabstop=4
|
||||
|
||||
# Copyright 2010 United States Government as represented by the
|
||||
# Administrator of the National Aeronautics and Space Administration.
|
||||
# All Rights Reserved.
|
||||
# Copyright 2012, 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
|
||||
# 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.
|
||||
"""
|
||||
Unit Tests for remote procedure calls using qpid
|
||||
"""
|
||||
|
||||
import mox
|
||||
|
||||
from nova import context
|
||||
from nova import log as logging
|
||||
from nova import test
|
||||
import nova.rpc.amqp as rpc_amqp
|
||||
|
||||
try:
|
||||
import qpid
|
||||
from nova.rpc import impl_qpid
|
||||
except ImportError:
|
||||
qpid = None
|
||||
impl_qpid = None
|
||||
|
||||
|
||||
LOG = logging.getLogger('nova.tests.rpc')
|
||||
|
||||
|
||||
class RpcQpidTestCase(test.TestCase):
|
||||
"""
|
||||
Exercise the public API of impl_qpid utilizing mox.
|
||||
|
||||
This set of tests utilizes mox to replace the Qpid objects and ensures
|
||||
that the right operations happen on them when the various public rpc API
|
||||
calls are exercised. The API calls tested here include:
|
||||
|
||||
nova.rpc.create_connection()
|
||||
nova.rpc.common.Connection.create_consumer()
|
||||
nova.rpc.common.Connection.close()
|
||||
nova.rpc.cast()
|
||||
nova.rpc.fanout_cast()
|
||||
nova.rpc.call()
|
||||
nova.rpc.multicall()
|
||||
"""
|
||||
|
||||
def setUp(self):
|
||||
self.mocker = mox.Mox()
|
||||
|
||||
self.mock_connection = None
|
||||
self.mock_session = None
|
||||
self.mock_sender = None
|
||||
self.mock_receiver = None
|
||||
|
||||
if qpid:
|
||||
self.orig_connection = qpid.messaging.Connection
|
||||
self.orig_session = qpid.messaging.Session
|
||||
self.orig_sender = qpid.messaging.Sender
|
||||
self.orig_receiver = qpid.messaging.Receiver
|
||||
qpid.messaging.Connection = lambda *_x, **_y: self.mock_connection
|
||||
qpid.messaging.Session = lambda *_x, **_y: self.mock_session
|
||||
qpid.messaging.Sender = lambda *_x, **_y: self.mock_sender
|
||||
qpid.messaging.Receiver = lambda *_x, **_y: self.mock_receiver
|
||||
|
||||
super(RpcQpidTestCase, self).setUp()
|
||||
|
||||
def tearDown(self):
|
||||
if qpid:
|
||||
qpid.messaging.Connection = self.orig_connection
|
||||
qpid.messaging.Session = self.orig_session
|
||||
qpid.messaging.Sender = self.orig_sender
|
||||
qpid.messaging.Receiver = self.orig_receiver
|
||||
|
||||
self.mocker.ResetAll()
|
||||
|
||||
super(RpcQpidTestCase, self).tearDown()
|
||||
|
||||
@test.skip_if(qpid is None, "Test requires qpid")
|
||||
def test_create_connection(self):
|
||||
self.mock_connection = self.mocker.CreateMock(self.orig_connection)
|
||||
self.mock_session = self.mocker.CreateMock(self.orig_session)
|
||||
|
||||
self.mock_connection.opened().AndReturn(False)
|
||||
self.mock_connection.open()
|
||||
self.mock_connection.session().AndReturn(self.mock_session)
|
||||
self.mock_connection.close()
|
||||
|
||||
self.mocker.ReplayAll()
|
||||
|
||||
connection = impl_qpid.create_connection()
|
||||
connection.close()
|
||||
|
||||
self.mocker.VerifyAll()
|
||||
|
||||
def _test_create_consumer(self, fanout):
|
||||
self.mock_connection = self.mocker.CreateMock(self.orig_connection)
|
||||
self.mock_session = self.mocker.CreateMock(self.orig_session)
|
||||
self.mock_receiver = self.mocker.CreateMock(self.orig_receiver)
|
||||
|
||||
self.mock_connection.opened().AndReturn(False)
|
||||
self.mock_connection.open()
|
||||
self.mock_connection.session().AndReturn(self.mock_session)
|
||||
if fanout:
|
||||
# The link name includes a UUID, so match it with a regex.
|
||||
expected_address = mox.Regex(r'^impl_qpid_test_fanout ; '
|
||||
'{"node": {"x-declare": {"auto-delete": true, "durable": '
|
||||
'false, "type": "fanout"}, "type": "topic"}, "create": '
|
||||
'"always", "link": {"x-declare": {"auto-delete": true, '
|
||||
'"exclusive": true, "durable": false}, "durable": true, '
|
||||
'"name": "impl_qpid_test_fanout_.*"}}$')
|
||||
else:
|
||||
expected_address = ('nova/impl_qpid_test ; {"node": {"x-declare": '
|
||||
'{"auto-delete": true, "durable": true}, "type": "topic"}, '
|
||||
'"create": "always", "link": {"x-declare": {"auto-delete": '
|
||||
'true, "exclusive": false, "durable": false}, "durable": '
|
||||
'true, "name": "impl_qpid_test"}}')
|
||||
self.mock_session.receiver(expected_address).AndReturn(
|
||||
self.mock_receiver)
|
||||
self.mock_receiver.capacity = 1
|
||||
self.mock_connection.close()
|
||||
|
||||
self.mocker.ReplayAll()
|
||||
|
||||
connection = impl_qpid.create_connection()
|
||||
connection.create_consumer("impl_qpid_test",
|
||||
lambda *_x, **_y: None,
|
||||
fanout)
|
||||
connection.close()
|
||||
|
||||
self.mocker.VerifyAll()
|
||||
|
||||
@test.skip_if(qpid is None, "Test requires qpid")
|
||||
def test_create_consumer(self):
|
||||
self._test_create_consumer(fanout=False)
|
||||
|
||||
@test.skip_if(qpid is None, "Test requires qpid")
|
||||
def test_create_consumer_fanout(self):
|
||||
self._test_create_consumer(fanout=True)
|
||||
|
||||
def _test_cast(self, fanout):
|
||||
self.mock_connection = self.mocker.CreateMock(self.orig_connection)
|
||||
self.mock_session = self.mocker.CreateMock(self.orig_session)
|
||||
self.mock_sender = self.mocker.CreateMock(self.orig_sender)
|
||||
|
||||
self.mock_connection.opened().AndReturn(False)
|
||||
self.mock_connection.open()
|
||||
self.mock_connection.session().AndReturn(self.mock_session)
|
||||
if fanout:
|
||||
expected_address = ('impl_qpid_test_fanout ; '
|
||||
'{"node": {"x-declare": {"auto-delete": true, '
|
||||
'"durable": false, "type": "fanout"}, '
|
||||
'"type": "topic"}, "create": "always"}')
|
||||
else:
|
||||
expected_address = ('nova/impl_qpid_test ; {"node": {"x-declare": '
|
||||
'{"auto-delete": true, "durable": false}, "type": "topic"}, '
|
||||
'"create": "always"}')
|
||||
self.mock_session.sender(expected_address).AndReturn(self.mock_sender)
|
||||
self.mock_sender.send(mox.IgnoreArg())
|
||||
# This is a pooled connection, so instead of closing it, it gets reset,
|
||||
# which is just creating a new session on the connection.
|
||||
self.mock_session.close()
|
||||
self.mock_connection.session().AndReturn(self.mock_session)
|
||||
|
||||
self.mocker.ReplayAll()
|
||||
|
||||
try:
|
||||
ctx = context.RequestContext("user", "project")
|
||||
|
||||
if fanout:
|
||||
impl_qpid.fanout_cast(ctx, "impl_qpid_test",
|
||||
{"method": "test_method", "args": {}})
|
||||
else:
|
||||
impl_qpid.cast(ctx, "impl_qpid_test",
|
||||
{"method": "test_method", "args": {}})
|
||||
|
||||
self.mocker.VerifyAll()
|
||||
finally:
|
||||
if rpc_amqp.ConnectionContext._connection_pool.free():
|
||||
# Pull the mock connection object out of the connection pool so
|
||||
# that it doesn't mess up other test cases.
|
||||
rpc_amqp.ConnectionContext._connection_pool.get()
|
||||
|
||||
@test.skip_if(qpid is None, "Test requires qpid")
|
||||
def test_cast(self):
|
||||
self._test_cast(fanout=False)
|
||||
|
||||
@test.skip_if(qpid is None, "Test requires qpid")
|
||||
def test_fanout_cast(self):
|
||||
self._test_cast(fanout=True)
|
||||
|
||||
def _test_call(self, multi):
|
||||
self.mock_connection = self.mocker.CreateMock(self.orig_connection)
|
||||
self.mock_session = self.mocker.CreateMock(self.orig_session)
|
||||
self.mock_sender = self.mocker.CreateMock(self.orig_sender)
|
||||
self.mock_receiver = self.mocker.CreateMock(self.orig_receiver)
|
||||
|
||||
self.mock_connection.opened().AndReturn(False)
|
||||
self.mock_connection.open()
|
||||
self.mock_connection.session().AndReturn(self.mock_session)
|
||||
rcv_addr = mox.Regex(r'^.*/.* ; {"node": {"x-declare": {"auto-delete":'
|
||||
' true, "durable": true, "type": "direct"}, "type": '
|
||||
'"topic"}, "create": "always", "link": {"x-declare": '
|
||||
'{"auto-delete": true, "exclusive": true, "durable": '
|
||||
'false}, "durable": true, "name": ".*"}}')
|
||||
self.mock_session.receiver(rcv_addr).AndReturn(self.mock_receiver)
|
||||
self.mock_receiver.capacity = 1
|
||||
send_addr = ('nova/impl_qpid_test ; {"node": {"x-declare": '
|
||||
'{"auto-delete": true, "durable": false}, "type": "topic"}, '
|
||||
'"create": "always"}')
|
||||
self.mock_session.sender(send_addr).AndReturn(self.mock_sender)
|
||||
self.mock_sender.send(mox.IgnoreArg())
|
||||
|
||||
self.mock_session.next_receiver().AndReturn(self.mock_receiver)
|
||||
self.mock_receiver.fetch().AndReturn(qpid.messaging.Message(
|
||||
{"result": "foo", "failure": False, "ending": False}))
|
||||
if multi:
|
||||
self.mock_session.next_receiver().AndReturn(self.mock_receiver)
|
||||
self.mock_receiver.fetch().AndReturn(
|
||||
qpid.messaging.Message(
|
||||
{"result": "bar", "failure": False,
|
||||
"ending": False}))
|
||||
self.mock_session.next_receiver().AndReturn(self.mock_receiver)
|
||||
self.mock_receiver.fetch().AndReturn(
|
||||
qpid.messaging.Message(
|
||||
{"result": "baz", "failure": False,
|
||||
"ending": False}))
|
||||
self.mock_session.next_receiver().AndReturn(self.mock_receiver)
|
||||
self.mock_receiver.fetch().AndReturn(qpid.messaging.Message(
|
||||
{"failure": False, "ending": True}))
|
||||
self.mock_session.close()
|
||||
self.mock_connection.session().AndReturn(self.mock_session)
|
||||
|
||||
self.mocker.ReplayAll()
|
||||
|
||||
try:
|
||||
ctx = context.RequestContext("user", "project")
|
||||
|
||||
if multi:
|
||||
method = impl_qpid.multicall
|
||||
else:
|
||||
method = impl_qpid.call
|
||||
|
||||
res = method(ctx, "impl_qpid_test",
|
||||
{"method": "test_method", "args": {}})
|
||||
|
||||
if multi:
|
||||
self.assertEquals(list(res), ["foo", "bar", "baz"])
|
||||
else:
|
||||
self.assertEquals(res, "foo")
|
||||
|
||||
self.mocker.VerifyAll()
|
||||
finally:
|
||||
if rpc_amqp.ConnectionContext._connection_pool.free():
|
||||
# Pull the mock connection object out of the connection pool so
|
||||
# that it doesn't mess up other test cases.
|
||||
rpc_amqp.ConnectionContext._connection_pool.get()
|
||||
|
||||
@test.skip_if(qpid is None, "Test requires qpid")
|
||||
def test_call(self):
|
||||
self._test_call(multi=False)
|
||||
|
||||
@test.skip_if(qpid is None, "Test requires qpid")
|
||||
def test_multicall(self):
|
||||
self._test_call(multi=True)
|
||||
|
||||
|
||||
#
|
||||
#from nova.tests.rpc import common
|
||||
#
|
||||
# Qpid does not have a handy in-memory transport like kombu, so it's not
|
||||
# terribly straight forward to take advantage of the common unit tests.
|
||||
# However, at least at the time of this writing, the common unit tests all pass
|
||||
# with qpidd running.
|
||||
#
|
||||
# class RpcQpidCommonTestCase(common._BaseRpcTestCase):
|
||||
# def setUp(self):
|
||||
# self.rpc = impl_qpid
|
||||
# super(RpcQpidCommonTestCase, self).setUp()
|
||||
#
|
||||
# def tearDown(self):
|
||||
# super(RpcQpidCommonTestCase, self).tearDown()
|
||||
#
|
||||
Reference in New Issue
Block a user