add notifier publisher for events

this allows the ability for deployers to push events to MQ rather
than send data directly to a database. the notifier publisher
supports the same options as the sample publisher.

Change-Id: Ia2cdfe351dad2336bce8519da3e568f226a2593a
Implements: blueprint notification-pipelines

1 files changed, 35 insertions, 15 deletions

diff --git a/ceilometer/publisher/messaging.py b/ceilometer/publisher/messaging.py
index 122ac24d..2e4d69fc 100644
--- a/ceilometer/publisher/messaging.py
+++ b/ceilometer/publisher/messaging.py
@@ -24,7 +24,6 @@ from oslo_config import cfg
 import six
 import six.moves.urllib.parse as urlparse
 
-import ceilometer
 from ceilometer.i18n import _
 from ceilometer import messaging
 from ceilometer.openstack.common import log
@@ -48,6 +47,11 @@ NOTIFIER_OPTS = [
                help='The topic that ceilometer uses for metering '
                'notifications.',
                ),
+    cfg.StrOpt('event_topic',
+               default='event',
+               help='The topic that ceilometer uses for event '
+               'notifications.',
+               ),
     cfg.StrOpt('metering_driver',
                default='messagingv2',
                help='The driver that ceilometer uses for metering '
@@ -102,7 +106,6 @@ class MessagingPublisher(publisher.PublisherBase):
                 cfg.CONF.publisher.metering_secret)
             for sample in samples
         ]
-
         topic = cfg.CONF.publisher_rpc.metering_topic
         self.local_queue.append((context, topic, meters))
 
@@ -142,18 +145,18 @@ class MessagingPublisher(publisher.PublisherBase):
 
     def _process_queue(self, queue, policy):
         while queue:
-            context, topic, meters = queue[0]
+            context, topic, data = queue[0]
             try:
-                self._send(context, topic, meters)
+                self._send(context, topic, data)
             except oslo.messaging.MessageDeliveryFailure:
-                samples = sum([len(m) for __, __, m in queue])
+                data = sum([len(m) for __, __, m in queue])
                 if policy == 'queue':
-                    LOG.warn(_("Failed to publish %d samples, queue them"),
-                             samples)
+                    LOG.warn(_("Failed to publish %d datapoints, queue them"),
+                             data)
                     return queue
                 elif policy == 'drop':
-                    LOG.warn(_("Failed to publish %d samples, dropping them"),
-                             samples)
+                    LOG.warn(_("Failed to publish %d datapoints, "
+                               "dropping them"), data)
                     return []
                 # default, occur only if rabbit_max_retries > 0
                 raise
@@ -167,7 +170,12 @@ class MessagingPublisher(publisher.PublisherBase):
         :param context: Execution context from the service or RPC call
         :param events: events from pipeline after transformation
         """
-        raise ceilometer.NotImplementedError
+        ev_list = [utils.message_from_event(
+            event, cfg.CONF.publisher.metering_secret) for event in events]
+
+        topic = cfg.CONF.publisher_notifier.event_topic
+        self.local_queue.append((context, topic, ev_list))
+        self.flush()
 
     @abc.abstractmethod
     def _send(self, context, topic, meters):
@@ -192,16 +200,28 @@ class RPCPublisher(MessagingPublisher):
 
 
 class NotifierPublisher(MessagingPublisher):
-    def __init__(self, parsed_url):
+    def __init__(self, parsed_url, topic):
         super(NotifierPublisher, self).__init__(parsed_url)
         self.notifier = oslo.messaging.Notifier(
             messaging.get_transport(),
             driver=cfg.CONF.publisher_notifier.metering_driver,
-            publisher_id='metering.publisher.%s' % cfg.CONF.host,
-            topic=cfg.CONF.publisher_notifier.metering_topic,
+            publisher_id='telemetry.publisher.%s' % cfg.CONF.host,
+            topic=topic,
             retry=self.retry
         )
 
-    def _send(self, context, event_type, meters):
+    def _send(self, context, event_type, data):
         self.notifier.sample(context.to_dict(), event_type=event_type,
-                             payload=meters)
+                             payload=data)
+
+
+class SampleNotifierPublisher(NotifierPublisher):
+    def __init__(self, parsed_url):
+        super(SampleNotifierPublisher, self).__init__(
+            parsed_url, cfg.CONF.publisher_notifier.metering_topic)
+
+
+class EventNotifierPublisher(NotifierPublisher):
+    def __init__(self, parsed_url):
+        super(EventNotifierPublisher, self).__init__(
+            parsed_url, cfg.CONF.publisher_notifier.event_topic)