projectweekend/RMQ-Utils

Automated RabbitMQ account and connection management.

This is a work in progress. I had a need for some services to automate admin tasks in RabbitMQ, so I decided to start building some things.

Account Creator

This is a class that makes one connection to the Rabbit Admin API and one to a direct exchange/queue named: rmq_utils/account_creator. It listens for messages, creates a new user/password/vhost via the Rabbit Admin API, and calls a post_create method for each. The message must be a JSON object:

{
    "user_key": "whatever"
}

The post_create method is implemented in your sub-class. It will receive two input parameters:

• user_key - The same value received in the message
• credentials - A dictionary representing the Rabbit account info for the account that was just created. It has the following keys: username, password, vhost

Example:

from rmq_utils import AccountCreator


RABBIT_URL = 'rabbit-url-to-listen-for-messages'
ROUTING_KEY = 'unique-key-for-server'
ADMIN_HOST = 'rabbit-admin-host'
ADMIN_USER = 'admin-user'
ADMIN_PASSWORD = 'admin-password'


class ExampleAccountCreator(AccountCreator):

    @staticmethod
    def post_create(user_key, creds):
        # Do whatever you want here, perhaps make a call to
        # update a record in a database using the user_key
        print('{0}: {1}'.format(user_key, creds['username']))


def main():
    creator = ExampleAccountCreator(
        rabbit_url=RABBIT_URL,
        routing_key=ROUTING_KEY,
        mgmt_host=ADMIN_HOST,
        mgmt_user=ADMIN_USER,
        mgmt_password=ADMIN_PASSWORD)
    try:
        creator.run()
    except KeyboardInterrupt:
        creator.stop()


if __name__ == '__main__':
    main()

Account Destroyer

This is a class that makes one connection to the Rabbit Admin API and one to a direct exchange/queue named: rmq_utils/account_destroyer. It listens for messages, deletes an existing user/vhost via the Rabbit Admin API, and calls a post_delete method for each. The message must be a JSON object:

{
    "user_key": "whatever",
    "rabbit_user": "the_username"
}

The post_delete method is implemented in your sub-class. It will receive one input parameter:

• user_key - The same value received in the message

Example:

from rmq_utils import AccountDestroyer


RABBIT_URL = 'rabbit-url-to-listen-for-messages'
ROUTING_KEY = 'unique-key-for-server'
ADMIN_HOST = 'rabbit-admin-host'
ADMIN_USER = 'admin-user'
ADMIN_PASSWORD = 'admin-password'


class ExampleAccountDestroyer(AccountDestroyer):

    @staticmethod
    def post_delete(user_key):
        # Do whatever you want here, perhaps make a call to
        # update a record in a database using the user_key
        print('{0}'.format(user_key))


def main():
    destroyer = ExampleAccountDestroyer(
        rabbit_url=RABBIT_URL,
        routing_key=ROUTING_KEY,
        mgmt_host=ADMIN_HOST,
        mgmt_user=ADMIN_USER,
        mgmt_password=ADMIN_PASSWORD)
    try:
        destroyer.run()
    except KeyboardInterrupt:
        destroyer.stop()


if __name__ == '__main__':
    main()

Connection Limiter

This service maintains a map of usernames to an array of connection names. It calls a user-defined method (max_connections_by_user) that returns a map of usernames to integers representing the maximum number of concurrent connections allowed for each. It compares the current number of open connections for each to the max, if above the maximum it closes connections to maintain the maximum.

Example:

from rmq_utils import ConnectionLimiter


class ExampleConnectionLimiter(ConnectionLimiter):

    @staticmethod
    def max_connections_by_user():
        # Do whatever you want here, perhaps make a call to
        # a database to get the max connection per user
        return {
            'test': 1
        }

def main():
    limiter = ExampleConnectionLimiter(
        mgmt_host=ADMIN_HOST,
        mgmt_user=ADMIN_USER,
        mgmt_password=ADMIN_PASSWORD)
    try:
        limiter.run()
    except KeyboardInterrupt:
        pass


if __name__ == '__main__':
    main()