Open Stack

Hey y'all!

Running into an interesting little dilemma with a branch I'm working on.
Recently, I introduced a branch in oslo-common to optionally .reject() a
kombu message on an exception. Currently, we always .ack() all messages
even if the processing callback fails. For Ceilometer, this is a problem
... we have to guarantee we get all notifications.

The patch itself was pretty simple, but didn't work :) The spawn_n()
call was eating the exceptions coming from the callback. So, in order to
get the exceptions it's simple enough to re-wrap the callback, but I
need to pool.waitall() after the spawn_n() to ensure none of the
consumers failed. Sad, but a necessary evil. And remember, it's only
used in a special case, normal openstack rpc is unaffected and remains
async.

But it does introduce a larger problem ... I have to change the rpc
callback signature.

Old: callback(message)
New: callback(message, delivery_info=None, wait_for_consumers=False)

(The delivery_info is another thing, we were dumping the message info on
the floor, but this has important info in it)

My worry is busting all the other callbacks out there that use
olso-common.rpc

Some options:
1. embed all these flags and extra data in the message structure

message = {'_context_stuff': "...",
           'payload": {...},
           '_extra_magic': {...}}

2. make a generic CallContext() object to include with message that has
anything else we need (a one-time signature break)

call_context = CallContext({"delivery_info": {...}, "wait": False})
callback(message, call_context)

3. some other ugly python hack that I haven't thought of yet.

Look forward to your thoughts on a solution!

Thanks
-S


My work-in-progess is here:
https://github.com/SandyWalsh/openstack-common/blob/callback_exceptions/openstack/common/rpc/amqp.py#L373