OpenStack uses RabbitMQ as a messaging system. There are several exchanges and queues for this purpose. I found that exchange named "nova" of type "topic" used for message transfer. Exchange use routing key to route message to queues (http://www.rabbitmq.com/tutorials/amqp-concepts.html).
(Useful image at http://www.rabbitmq.com/img/tutorials/intro/hello-world-example-routing.png - not enough reputation to post it here)
There are several queues in OpenStack like compute, cert, network and so on. They use routing key with the same name. So I created several new queues with these routing keys to bind them with consumer that handle messages. For example, there is queue named "compute" that use routing key named "compute". I created new queue "my_compute" that use same routing key. As I think it should work and I will get messages.
I have some code that connects to exchange, creates my queues and the consumer.
def connect(params):
connection = kombu.Connection(hostname=params['host'])
exchange = kombu.entity.Exchange(name=params['exchange_name'],
type=params['exchange_type'],
durable=params['exchange_durable'],
auto_delete=params['exchange_auto_delete'],
internal=params['exchange_internal'])
queue_list = []
for queue in params['queues_params']:
queue_list.append(kombu.messaging.Queue(name=queue['name'],
exchange=exchange,
routing_key=queue['routing_key'],
channel=connection.channel(),
durable=queue['durable'],
auto_delete=queue['auto_delete']))
consumer = kombu.messaging.Consumer(channel=connection.channel(),
queues=queue_list,
no_ack=True,
callbacks=[self._process_message])
consumer.consume()
return connection
Argument "params" is the map that got from json file:
{
"host" : "xxx",
"exchange_name" : "nova",
"exchange_type" : "topic",
"exchange_durable" : false,
"exchange_auto_delete" : false,
"exchange_internal" : false,
"queues_params" : [
{
"name" : "my_compute",
"routing_key" : "compute",
"durable" : false,
"auto_delete" : false,
"arguments" : [ ]
},
{
"name" : "my_network",
"routing_key" : "network",
"durable" : false,
"auto_delete" : false,
"arguments" : [ ]
},
.
.
.
It's working. But I only get messages for network queue . I don't know are there any other messages, but it looks like there are. Am I right? Or something is wrong? Is there other messages and how can I get them?
heckj :
This code is undergoing some active changes in this development cycle, but at the moment I'd assert you're looking a bit deeply. For most of the nova components, the queue interfaces are abstracted away underneath the RPC common library that those components use, and that chooses the topics and queues.\n\nIn particular, topics can also be host specific when the RPC code wants to send a message to a specific compute, network, or storage host. The only messages you'll be seeing above are the general broadcast messages, which in practice will often be requests for information about floating IP addresse\n\nIf you want an example of something that snags all the messages on the queue, you should definitely look at Ceilometer, which is intended to do just that, and leverages the notification system that's also embedded into nova and related components. It's not going to offer you the same thing that intercepting and interpretting nova, network, and cinder messages would - just sort of depends on your overall goals if that's useful.",
2013-05-27T19:05:19