When a caller msg doesn't exists anymore but a reply is ready for it.
amqp driver drop the whole message into a logging.warn. That can be a
bit huge in some case.
This change just writes the message id to the WARN level and put the queues
and the messages into the debug level.
Change-Id: Ibcc6b066171cdea48f102ca1bd85f81c639fbbb5
If you don't close the AMQP connection, then the connection
remains open and the next time when the messages are sent on
the listening topic, then some of the messages will not be processed
as there is no dispatcher running to process the message.
Closes-Bug: #1335086
Change-Id: I1f39eedf1500b6b6209ae0222f32e08e304895e0
Cast messages will not contain a msg_id and we're currently
sending messages to the default excahgne with empty msg_id.
In order to fix it we should not send replies if there is no
msg_id in message.
Change-Id: I5b1142029d2c718c3929cf6cf1f6e958b95a5c96
Closes-bug: #1355058
Passing mutable objects as default args is a known Python pitfall.
We'd better avoid this.
Change-Id: I67cc0774a65886ef9fce0b72e52157b622248a85
Closes-Bug: #1327473
There are files containing string format arguments inside
logging messages. Using logging function parameters should
be preferred.
Change-Id: I4a7ce6916aefb2f2d445f0ebd977c824b1c51e24
Partial-Bug: #1321274
When a rpc client try to make a RPC call and the server is unreachable
The rpc call hang until the server come back.
In most case this is the desired behavior.
But sometimes, we can prefer that the library raise an exception after a
certain number of retries.
For example in ceilometer, when publishing a
storage.objects.incoming.bytes sample from the Swift middleware to an
AMQP topic, you might not want to block the Swift client if the AMQP broker
is unavailable - instead, you might have a queueing policy whereby
if a single reconection attempt fails we queue the sample in memory and
try again when another sample is to be published.
This patch is the oslo.messaging part that allow this.
Closes bug #1282639
Co-Authored-By: Ala Rezmerita <ala.rezmerita@cloudwatt.com>
Change-Id: I32086d0abf141c368343bf225d4b021da496c020
For asynchronous programming, a timeout parameter is required on the listener
to allow to stop it at exit. poll() returns None on timeout.
It plan to use it in my new asyncio (Trollius) executor:
https://review.openstack.org/#/c/70983/
See also the related blueprint for the rationale:
https://wiki.openstack.org/wiki/Oslo/blueprints/asyncio
Change-Id: I918ae3c267743a0eaed1d6a210c79fb4a0eb8733
This change remove the hack to set the default exchange of a transport in the
amqp driver, by removing the usage of the configuration object to get the
default exchange in rabbit and qpid driver, and instead use the value
passed to the driver constructor into all amqp publishers and consumers
class/method that needs it.
Closes-bug: #1256345
Change-Id: Iba54ca79a49f8545854205c1451b2403735c1006
gettextutils is expecting to receive unicode strings
rather than basestrings. A basestring can cause an
unhandled exception in the logging code. To help avoid
such issues we should remove str() from LOG.* messages and
exceptions. We have verified that the %s formatting code
properly handle getting strings to unicode where necessary.
Copied from https://review.openstack.org/#/c/77722
Change-Id: I082af5c9ae8bf9859382c2c387b10b48358e10b3
Related-Bug: #1286306
This patch add the support of multiple hosts in transport url for rabbit and
qpid drivers. And also fix the amqp connection pool management to allow
to have one pool by transport.
Implements blueprint multiple-hosts-support-in-url
Co-Authored-By: Ala Rezmerita <ala.rezmerita@cloudwatt.com>
Change-Id: I5aff24d292b67a7b65e33e7083e245efbbe82024
Drivers have a notify_send() method which allows for special semantics
to be implemented for notification sending. During the port over to
oslo.messaging, it appears we stopped using this and switched to just
using topic_send() instead.
Lets restore the use of notify_send() so that we don't lose behaviours
like I9f8dfd6eaf3996be58fecff6ad91508bdcef23f3 added.
Change-Id: Id0ad0431d3e2b1bfe21723e7f3b1842c2d3a9546
In commit d8d2ad9 we delayed when a message's unique ID gets added to
the duplicate message cache in order to allow for message requeueing.
However, as part of this, we exposed the private _unique_id field
outside of the driver. This commit reverses that change by storing
the ID in the AMQPIncomingMessage object.
Change-Id: Ibeb7896de7ad9abf3c6a43495c1a87aabb762c0d
This patch allow to requeue the notification received by the
notification listener.
Partial implements blueprint notification-subscriber-server
Change-Id: I49c4ba91224c280e479edb19289ccb337a2ab843
This patch make the dispatcher responsible of the message
acknowledgement.
This is the preliminar step to be able to requeue message.
Partial implements blueprint notification-subscriber-server
Change-Id: If74b47d5e23976d407deb27df7395b1982963c75
The patch add ta abstraction layer to acknowledge a message.
Partial implements blueprint notification-subscriber-server
Change-Id: I6e37780cc28737cfd56b6719ec8d9cebbc9bb278
This patch allow the fake driver to comsume multiple topics
with one listener.
Partial implements blueprint notification-subscriber-server
Change-Id: Ib52dc181e10b487854fbb398eda9f758232a1251
On the client side, in the rabbit and qpid drivers, we use a connection
pool to avoid opening a connection for each message we send. However,
there is only currently one connection pool per process:
def get_connection_pool(conf, connection_cls):
with _pool_create_sem:
# Make sure only one thread tries to create the connection pool.
if not connection_cls.pool:
connection_cls.pool = ConnectionPool(conf, connection_cls)
return connection_cls.pool
This is a nasty artifact of the original RPC having no conectp of a
transport context - everything was a global. We'll fix this soon enough.
In the meantime, we need to make sure we only use this connection pool
where we're not using the default transport configuration from the
config file - i.e. where we supply a transport URL.
The use case here is cells - we send messages to a remote cell by
connecting to it using a transport URL. In our devstack testing, the
two cells are on the same Rabbit broker but under different virtual
hosts. Because we were always using the connection pool on the client
side, we were seeing both cells always send messages to the '/' virtual
host.
Note - avoiding the connection pool in the case of cells is the same
behaviour as the current RPC code:
def cast_to_server(conf, context, server_params, topic, msg, connection_pool):
...
with ConnectionContext(conf, connection_pool, pooled=False,
server_params=server_params) as conn:
Change-Id: I2f35b45ef237bb85ab8faf58a408c03fcb1de9d7
Currently, if we are supplied with a transport URL with only the virtual
host specified, we completely ignore it. Instead, the behaviour should
be that we use that virtual host with the host, port and credentials
from the config file.
Change-Id: Ic97aa511ddf9bce69b1a5069d9f6468f4bd6dd4c
Concurrency. Sigh.
A sequence of events like this is possible:
- We send a request from thread A
- Thread B, who is waiting for a response gets scheduled
- Thread B receives our response and queues it up
- Thread B receives its own response and drops the connection lock
- Thread A grabs the connection lock and wait for a response to arrive
The obvious solution is that when we grab the connection lock, we should
check whether a previous lock-holding thread had already received our
response and queued it up.
Change-Id: I88b0d55d5a40814a84d82ed4f42d5ba85d2ef7e0
On the server side, we only send replies if the request included a
_msg_id key. Also, the _reply_q key is only used when we wish to send a
reply.
So, in order to retain the exact same on-the-wire behaviour and ensure
servers aren't sending replies where none is needed, only include these
keys if we're doing a call (i.e. wait_for_reply=True).
Change-Id: Iac329493252be7d94b1ebe24f00e4d3f5c61d269
Just like the bug in the fake driver, we have a similar rather
embarassing and obvious bug - we're currently not allowing endpoint
methods to send replies of None, '', False, [], {}, etc.
Change-Id: Icf0abdfcf122c5757dd3737f66130b3a53769ef6
Nova's cells/rpc_driver.py has some code which allows user of the REST
API to update elements of a cell's transport URL (say, the host name of
the message broker) stored in the database. To achieve this, it has
a parse_transport_url() method which breaks the URL into its constituent
parts and an unparse_transport_url() which re-forms it again after
updating some of its parts.
This is all fine and, since it's fairly specialized, it wouldn't be a
big deal to leave this code in Nova for now ... except the unparse
method looks at CONF.rpc_backend to know what scheme to use in the
returned URL if now backend was specified.
oslo.messaging registers the rpc_backend option, but the ability to
reference any option registered by the library should not be relied upon
by users of the library. Imagine, for instance, if we renamed the option
in future (with backwards compat for old configurations), then this
would mean API breakage.
So, long story short - an API along these lines makes some sense, but
especially since not having it would mean we'd need to add some way to
query the name of the transport driver.
In this commit, we add a simple new TransportURL class:
>>> url = messaging.TransportURL.parse(cfg.CONF, 'foo:///')
>>> str(url), url
('foo:///', <TransportURL transport='foo'>)
>>> url.hosts.append(messaging.TransportHost(hostname='localhost'))
>>> str(url), url
('foo://localhost/', <TransportURL transport='foo', hosts=[<TransportHost hostname='localhost'>]>)
>>> url.transport = None
>>> str(url), url
('kombu://localhost/', <TransportURL transport='kombu', hosts=[<TransportHost hostname='localhost'>]>)
>>> cfg.CONF.set_override('rpc_backend', 'bar')
>>> str(url), url
('bar://localhost/', <TransportURL transport='bar', hosts=[<TransportHost hostname='localhost'>]>)
The TransportURL.parse() method equates to parse_transport_url() and
TransportURL.__str__() equates to unparse_transport().
The transport drivers are also updated to take a TransportURL as a
required argument, which simplifies the handling of transport URLs in
the drivers.
Change-Id: Ic04173476329858e4a2c2d2707e9d4aeb212d127
If a transport URL is supplied, transform it into the server_params
format which was previously used for cast_to_server() etc.
Change-Id: I453734a71748dc8d3ffc02ead7bfb92ffb0a6c7c
My original thinking was that if you're using the exchange name to
separate two instances of the applications (so their queues don't
collide) then the exchange name is pretty key to transport
configuration. In fact, it's really a virtual host that you'd use for
this (at least in the case of rabbit and qpid).
Also, Nova's cells code has already moved ahead with the assumption that
the path specifies a virtual host, so it'd only make sense to deviate
from that if there was a really good reason to.
Change-Id: Ic8b5dc3538b6b17afec524047acc2efa76366377
We shouldn't be logging expected exceptions. Add a failing rabbit driver
test to check this and fix it.
Change-Id: I78b758957117be7c11c5826a27dd6d1d4fffe9cb
Oslo's logging code has some useful support for including bits of the
request context in log messages. While this isn't exclusively about the
request context in a dispatching RPC method, it seems useful for
oslo.messaging to support the concept for at least this use case simply
by recording the context in a thread local store before dispatching an
endpoint method and immediately clearing it when the method returns.
Note, we don't need to store weak refs in our store because we will
clear the reference in all cases rather than ever leaving a stale
reference around in the store.
Change-Id: I70ac06ed3a2a891a7a7b388b1823a0f3b08f2dd1
Currently we have a allowed_rpc_exception_modules configuration variable
which we use to configure a per-project list of modules which we will
allow exceptions to be instantiated from when deserializing remote
errors.
It makes no sense for this to be user configurable, instead the list of
modules should be set when you create a transport.
Closes-Bug: #1031719
Change-Id: Ib40e92cb920996ec5e8f63d6f2cbd88fd01a90f2
Review I4e7b19dc730342091fd70a717065741d56da4555 gives a lot of the
background here, but the idea is that some exceptions raised by an RPC
endpoint method do not indicate any sort of failure and should not be
logged by the server as an error.
The classic example of this is conductor's instance_get() method raising
InstanceNotFound. This is perfectly normal and should not be considered
an error.
The new API is a decorator which you can use with RPC endpoints methods
to indicate which exceptions are expected:
@messaging.expected_exceptions(InstanceNotFound)
def instance_get(self, context, instance_id):
...
but we also need to expose the ExpectedException type itself so that
direct "local" users of the endpoint class know what type will be used
to wrap expected exceptions. For example, Nova has an ExceptionHelper
class which unwraps the original exception from an ExpectedException and
re-raises it.
I've changed from client_exceptions() and ClientException to make it
more clear it's intent. I felt that the "client" naming gave the
impression it was intended for use on the client side.
Change-Id: Ieec4600bd6b70cf31ac7925a98a517b84acada4d
Notifications are an unusual case in that we need users to manually opt
in to new incompatible message formats by editing configuration because
there may be external consumers expecting the old format.
Add a send_notification() method to the driver interface and add a
format version paramater to the method, to make it clear that this
version selection is specifically for notifications.
In the case of the rabbit/qpid drivers, the 2.0 format is where we added
the message envelope.
Change-Id: Ib4925c308b1252503749962aa16f043281f2b429
The target preconditions (e.g. you need at least a topic to send) are
the same for all drivers, so enforce them before we ever call into a
driver.
Change-Id: Ic4e9bd94bd9f060ec0662d2bb778c699903dddc4
This method is fairly gnarly, so break my usual preference to make the
intent clear from the code and instead include detailed comments in the
method.
Change-Id: I107272a7eab85c70581652488a3c14ce0e18b906
These FIXMEs don't need fixing now that I think it through some more.
The reply format is specific to drivers and the 'ending' flag is part of
the existing wire protocol that we need to support even if we don't
support multicall().
Change-Id: I834f0bb01513b5318f0b365948a7d9247feb49bf
We appear to not have a use for this. I had originally thought we might
use this to ack messages one they've been processed and replied to, but
we actually have always acked messages as soon as they have been
deserialized and queued for dispatching.
Change-Id: I8e1fd565814f3b5e3ba0f1bc77e62ed52ff08661
Make the rabbit driver properly serialize exceptions before sending them
back in a reply and then properly re-raise them on the client side.
Also, extend the rabbit driver test to cover this case.
Change-Id: I6b3d03edcd41810125ba6442db5515754f0c1ac9
The trickiest logic in the rabbit driver is to handle the situation
where multiple threads are waiting for a reply on the same reply queue.
This commit adds unit testing for that scenario and fixes some bugs with
it.
Change-Id: I5c8fbeec49572a4f3badbcdae414dc44dc690b6a
While we're iterating over the queues in ReplyWaiters.wake_all(), new
queues can be registered and we get:
RuntimeError: dictionary changed size during iteration
Instead of using an iterator, take a snapshot list of message IDs and
operate on that.
We don't actually care about any new queues added after wake_all() is
called because the connection lock has already been dropped so one of
the other waiters must have picked it up.
We also don't need to worry about queues being removed - if we write to
a removed queue, that's not going to be a problem.
Change-Id: Ib572cbfd3a7346b76579f82b64aa85a03c1a4fb2
