alerta_amqp.py

import logging
import os

from kombu import BrokerConnection, Exchange, Producer
from kombu.utils.debug import setup_logging

try:
    from alerta.plugins import app  # alerta >= 5.0
except ImportError:
    from alerta.app import app  # alerta < 5.0
from alerta.plugins import PluginBase

LOG = logging.getLogger('alerta.plugins.amqp')

DEFAULT_AMQP_URL = 'mongodb://localhost:27017/kombu'
DEFAULT_AMQP_TOPIC = 'notify'

AMQP_URL = os.environ.get('REDIS_URL') or os.environ.get('AMQP_URL') or app.config.get('AMQP_URL', DEFAULT_AMQP_URL)
AMQP_TOPIC = os.environ.get('AMQP_TOPIC') or app.config.get('AMQP_TOPIC', DEFAULT_AMQP_TOPIC)


class FanoutPublisher(PluginBase):

    def __init__(self, name=None):
        if app.config['DEBUG']:
            setup_logging(loglevel='DEBUG', loggers=[''])

        self.connection = BrokerConnection(AMQP_URL)
        try:
            self.connection.connect()
        except Exception as e:
            LOG.error('Failed to connect to AMQP transport %s: %s', AMQP_URL, e)
            raise RuntimeError

        self.channel = self.connection.channel()
        self.exchange_name = AMQP_TOPIC

        self.exchange = Exchange(name=self.exchange_name, type='fanout', channel=self.channel)
        self.producer = Producer(exchange=self.exchange, channel=self.channel)

        super(FanoutPublisher, self).__init__(name)

        LOG.info('Configured fanout publisher on topic "%s"', AMQP_TOPIC)

    def pre_receive(self, alert):
        return alert

    def post_receive(self, alert):
        LOG.info('Sending message %s to AMQP topic "%s"', alert.get_id(), AMQP_TOPIC)
        body = alert.get_body()
        LOG.debug('Message: %s', body)
        self.producer.publish(body, declare=[self.exchange], retry=True)

    def status_change(self, alert, status, text):
        return