Listening to notifications in Openstack

Another short one. As mostly anyone that has had to maintain an OpenStack cluster will know, many of the OpenStack services has the ability to report notifications via the message bus. These notifications will be enabled in most production deployments but DevStack doesn’t enable them by default. Fortunately enabling them is pretty painless. If we wanted to enable the notifications in, say, Keystone, we can do so by configuring the following in /etc/keystone/keystone.conf:

[oslo_messaging_notifications]
transport_url = rabbit://stackrabbit:***@***:5672/
driver = messagingv2

These options are provided by the oslo.messaging project, which virtually every OpenStack service harnesses.

Once you’ve enabled them though, you what can you do with them? Using the messagingv2 means they’re published over the AMQP message bus (and the message is enveloped, which is the difference to messaging), however, they’re not exposed via HTTP APIs or similar. We could whip out a telemetry tool like Ceilometer but that seems overkill for a quick bit of debugging. Well it turns out oslo.messaging provides utilities for both creating notifications and listening to them. Writing a listener is relatively simple:

import json
import logging

from oslo_config import cfg
import oslo_messaging

cfg.CONF()
logging.basicConfig(level=logging.INFO)
LOG = logging.getLogger()

URL = 'rabbit://stackrabbit:***@***:5672/'


class NotificationEndpoint:

    def _notify(self, ctxt, publisher_id, event_type, payload, metadata):
        LOG.info('notification received %s:%s' % (publisher_id, event_type))
        output = json.dumps(
            {
                "payload": payload,
                "publisher_id": publisher_id,
                "event_type": event_type,
            },
            indent=4,
        )
        print(output)

    audit = _notify
    debug = _notify
    info = _notify
    warn = _notify
    error = _notify
    critical = _notify
    sample = _notify


def main():
    transport = oslo_messaging.get_notification_transport(cfg.CONF, url=URL)
    targets = [oslo_messaging.Target(topic='notifications')]
    endpoints = [NotificationEndpoint()]

    server = oslo_messaging.get_notification_listener(
        transport,
        targets,
        endpoints,
        executor='threading',
    )

    LOG.info("messaging starting")
    server.start()
    LOG.info("messaging started")
    server.wait()
    LOG.info("exit")


if __name__ == '__main__':
    main()

If you run this and then, say, attempt to authenticate with invalid credentials, you’ll see notifications arriving:

{
    "payload": {
        "typeURI": "http://schemas.dmtf.org/cloud/audit/1.0/event",
        "eventType": "activity",
        "id": "68c5b24b-89f0-5c93-abe4-5c3d6651fb00",
        "eventTime": "2023-07-07T15:52:00.496156+0000",
        "action": "authenticate",
        "outcome": "failure",
        "observer": {
            "id": "08d95fde312344debbcc00106b139995",
            "typeURI": "service/security"
        },
        "initiator": {
            "id": "78fab588d9d34900a6c59fc22f3a2049",
            "typeURI": "service/security/account/user",
            "host": {
                "address": "***",
                "agent": "openstacksdk/1.3.0 keystoneauth1/5.2.1 python-requests/2.31.0 CPython/3.11.4"
            },
            "request_id": "req-0126be50-db3a-4975-a814-d717a5fdfdf8",
            "user_id": "***",
            "username": "***"
        },
        "target": {
            "id": "fc220ce7-2787-5152-bf81-9943cb8ee24b",
            "typeURI": "service/security/account/user"
        }
    },
    "publisher_id": "identity.stephenfin-devstack",
    "event_type": "identity.authenticate"
}

This can be extremely valuable for debugging since notifications will expose things that aren’t necessarily obvious from logs.