# Path: monasca_notification/common/utils.py
# def get_statsd_client(dimensions=None):
#     local_dims = dimensions.copy() if dimensions else {}
#     local_dims.update(NOTIFICATION_DIMENSIONS)
#     if CONF.statsd.enable:
#         LOG.debug("Establishing connection with statsd on {0}:{1}"
#                   .format(CONF.statsd.host, CONF.statsd.port))
#         client = monascastatsd.Client(name='monasca',
#                                       host=CONF.statsd.host,
#                                       port=CONF.statsd.port,
#                                       dimensions=local_dims)
#     else:
#         LOG.warn("StatsD monitoring disabled. Overriding monascastatsd.Client to use it offline")
#         client = OfflineClient(name='monasca',
#                                host=CONF.statsd.host,
#                                port=CONF.statsd.port,
#                                dimensions=local_dims)
#     return client
# 
# Path: monasca_notification/processors/alarm_processor.py
# CONF = cfg.CONF
# class AlarmProcessor(object):
#     def __init__(self):
#     def _parse_alarm(alarm_data):
#     def _alarm_is_valid(self, alarm):
#     def _build_notification(self, alarm):
#     def to_notification(self, raw_alarm):
# 
# Path: monasca_notification/processors/notification_processor.py
# class NotificationProcessor(object):
#     def __init__(self):
#     def _remaining_plugin_types(self):
#     def insert_configured_plugins(self):
#     def send(self, notifications):


import time
from oslo_config import cfg
from oslo_log import log as logging
from monasca_common.kafka import client_factory
from monasca_notification.common.utils import get_statsd_client
from monasca_notification.processors import alarm_processor as ap
from monasca_notification.processors import notification_processor as np
            CONF.kafka.url,
            CONF.kafka.group,
            CONF.kafka.alarm_topic,
            CONF.zookeeper.url,
            CONF.zookeeper.notification_path,
            CONF.kafka.legacy_kafka_client_enabled)
        self._producer = client_factory.get_kafka_producer(
            CONF.kafka.url,
            CONF.kafka.legacy_kafka_client_enabled)
        self._alarms = ap.AlarmProcessor()
        self._notifier = np.NotificationProcessor()

    def _add_periodic_notifications(self, notifications):
        for notification in notifications:
            period = str(notification.period)
            if period in CONF.kafka.periodic.keys():
                notification.notification_timestamp = time.time()
                self._producer.publish(CONF.kafka.periodic[period],
                                       [notification.to_json()])

    def run(self):
        finished_count = self._statsd.get_counter(name='alarms_finished_count')
        for alarm in self._consumer:
            log.debug('Received alarm >|%s|<', str(alarm))
            notifications, partition, offset = self._alarms.to_notification(alarm)
            if notifications:
                self._add_periodic_notifications(notifications)

                sent, failed = self._notifier.send(notifications)
                self._producer.publish(CONF.kafka.notification_topic,

# Path: monasca_notification/common/utils.py
# def get_statsd_client(dimensions=None):
#     local_dims = dimensions.copy() if dimensions else {}
#     local_dims.update(NOTIFICATION_DIMENSIONS)
#     if CONF.statsd.enable:
#         LOG.debug("Establishing connection with statsd on {0}:{1}"
#                   .format(CONF.statsd.host, CONF.statsd.port))
#         client = monascastatsd.Client(name='monasca',
#                                       host=CONF.statsd.host,
#                                       port=CONF.statsd.port,
#                                       dimensions=local_dims)
#     else:
#         LOG.warn("StatsD monitoring disabled. Overriding monascastatsd.Client to use it offline")
#         client = OfflineClient(name='monasca',
#                                host=CONF.statsd.host,
#                                port=CONF.statsd.port,
#                                dimensions=local_dims)
#     return client
# 
# Path: monasca_notification/processors/alarm_processor.py
# CONF = cfg.CONF
# class AlarmProcessor(object):
#     def __init__(self):
#     def _parse_alarm(alarm_data):
#     def _alarm_is_valid(self, alarm):
#     def _build_notification(self, alarm):
#     def to_notification(self, raw_alarm):
# 
# Path: monasca_notification/processors/notification_processor.py
# class NotificationProcessor(object):
#     def __init__(self):
#     def _remaining_plugin_types(self):
#     def insert_configured_plugins(self):
#     def send(self, notifications):


# Path: monasca_notification/notification_engine.py
import time
from oslo_config import cfg
from oslo_log import log as logging
from monasca_common.kafka import client_factory
from monasca_notification.common.utils import get_statsd_client
from monasca_notification.processors import alarm_processor as ap
from monasca_notification.processors import notification_processor as np

            CONF.kafka.url,
            CONF.kafka.group,
            CONF.kafka.alarm_topic,
            CONF.zookeeper.url,
            CONF.zookeeper.notification_path,
            CONF.kafka.legacy_kafka_client_enabled)
        self._producer = client_factory.get_kafka_producer(
            CONF.kafka.url,
            CONF.kafka.legacy_kafka_client_enabled)
        self._alarms = ap.AlarmProcessor()
        self._notifier = np.NotificationProcessor()

    def _add_periodic_notifications(self, notifications):
        for notification in notifications:
            period = str(notification.period)
            if period in CONF.kafka.periodic.keys():
                notification.notification_timestamp = time.time()
                self._producer.publish(CONF.kafka.periodic[period],
                                       [notification.to_json()])

    def run(self):
        finished_count = self._statsd.get_counter(name='alarms_finished_count')
        for alarm in self._consumer:
            log.debug('Received alarm >|%s|<', str(alarm))
            notifications, partition, offset = self._alarms.to_notification(alarm)
            if notifications:
                self._add_periodic_notifications(notifications)

                sent, failed = self._notifier.send(notifications)
                self._producer.publish(CONF.kafka.notification_topic,

                                       [i.to_json() for i in sent])