ODS Grizzly: Using the Message Bus for Messages

One of the sessions I am leading at the ODS this week is Using the Message Bus for Messages in the openstack-common track. I proposed this session because the current abstractions in openstack.common.rpc, used for communication between OpenStack components by sending messages over AMQP or ZMQ, do not meet the needs of the Ceilometer project. Below are my notes for the session, also available in the etherpad.

Background

The API includes methods for sending and receiving RPC calls and for sending notification messages. It does not include usable APIs for subscribing to notifications, or for sending and receiving generic messages to be consumed by multiple workers. Both Ceilometer and Quantum are hijacking notifications and using a low-level API to achieve generic messaging, but we need to add a public API and ensure that all of the RPC drivers support the pattern. The goal of this session is to agree on the basic design for such an API to be added during Grizzly.

Along with a way to receive notifications, we should also consider adding a more generic message API that is not tied to RPC. For example, as adoption of the ceilometer project grows, we may want to include a library for creating and sending metering messages from other services. We have such a library, using RPC right now, and it is mostly decoupled from the rest of ceilometer. At DreamHost we are building a service that will use the library, which will help us discover any other issues with removing it from ceilometer. It really shouldn’t need to be based on RPC, though, since the messages are one way and meant to be consumed but need no reply.

I am using AMQP terminology below. We can come up with more “generic” terms, but for the sake of discussion and so everyone understands what I mean, I’m sticking with the existing terms.

Proposal

• create openstack.common.messaging
• leave openstack.common.rpc alone, or rebuild it on top of the messaging API if someone wants

Requirements for openstack.common.messaging

• publishing
  • publish messages without knowing about the consumer (no method name for receiver)
  • minimal standard message envelope, with arbitrary JSON payload
  • much like notifications
  • message_id
  • publisher_id
  • timestamp
  • message_type (replaces event_type, same as routing key?)
  • message_type_version (or just include in type name?)
  • payload
  • signature (HMAC or other)
• message-class based exchanges instead of service-named exchanges
  • allows multiple publishers and subscribers to collaborate based on the type of the message (network, compute, notifications) instead of requiring subscribers to know the names of all possible publishers
  • the specific message type (the event_type from notification or some other value for new messages) acts as the routing key
  • will need to be able to specify the exchange and routing key explicitly for each outgoing message
• subscribing
  • subscribe to messages without interfering with other consumers
  • separate queues receiving messages with the same routing key or topic
  • subscribe to messages and collaborate with other consumers
  • shared queue, e.g., declare_worker() or declare_topic_consumer()
  • explicitly bind incoming topics to an arbitrary method for processing
  • method name shouldn’t be required in the message payload
  • multiple messages going through the same dispatcher (ceilometer collector does not know which notifications are needed in advance)

Nice to have

• object-based API instead of dictionaries on both sides (publish and subscribe)
• no module-level API accessing global config
• create an instance of the thing that does the work and use it

API Example

This straw man proposal for the publishing API is here to start the discussion. I like it, but I’m sure it can be improved.

class Message(object):
        def __init__(self, message_type, version):
                self.message_type = message_type
                self.version = version
        def __json__(self):
                return {...}

class ObjectToSend(Message):
        def __init__(self, arg1):
                super(ObjectToSend, self).__init__(
                        message_type='some.string',
                        version='1.0',
                        )
                self.arg1 = arg1

obj = ObjectToSend('blah')
e = Exchange(exchange_name)
mf = MessageFactory(signing_key)
m = mf.create_message(obj)  # uses obj.__json__() and other properties
e.send(routing_key, m)

The exchange and signing_key are probably coupled (different keys for different exchanges), so the Exchange and MessageFactory will be kept together. They are separated above to illustrate the responsibilities, but there could be an encapsulating Publisher object that knows how to do all of this work:

p = Publisher(e, mf)
p.send(routing_key, obj)

A Router could know about the different types of messages used within an application:

class Router(object):
        def __init__(self, publisher):
                self.publisher = publisher
        def send(self, routing_key, obj):
                self.publisher.send(routing_key, obj)

class CeilometerRouter(Router):
        def send_meter_message(self, meter):
                self.send('metering', meter)

r = CeilometerRouter(p)
r.send_meter_message(Counter(...))

• messaging
 
• OpenStack