Add listener service for event alarm evaluation

This patch adds an event listener service which listens event queue and
triggers evaluations of 'event' type alarms.

DocImpact

Change-Id: I5ee00ab50a0181bacee6d037ce61f40c1eddcfa2
Implements: blueprint event-alarm-evaluator

aodh/cmd/eventlet/alarm.py

@@ -18,6 +18,7 @@
 from oslo_service import service as os_service
 
 from aodh import evaluator as evaluator_svc
+from aodh import event as event_svc
 from aodh import notifier as notifier_svc
 from aodh import service
 
@@ -30,3 +31,8 @@ def notifier():
 def evaluator():
     conf = service.prepare_service()
     os_service.launch(conf, evaluator_svc.AlarmEvaluationService(conf)).wait()
+
+
+def listener():
+    conf = service.prepare_service()
+    os_service.launch(conf, event_svc.EventAlarmEvaluationService(conf)).wait()

aodh/event.py

@@ -0,0 +1,66 @@
+#
+# Copyright 2015 NEC Corporation.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+from oslo_config import cfg
+import oslo_messaging
+from oslo_service import service
+
+from aodh.evaluator import event
+from aodh import messaging
+from aodh import rpc
+from aodh import storage
+
+
+OPTS = [
+    cfg.StrOpt('event_alarm_topic',
+               default='alarm.all',
+               help='The topic that aodh uses for event alarm evaluation.'),
+]
+
+
+class EventAlarmEndpoint(object):
+
+    def __init__(self, evaluator):
+        self.evaluator = evaluator
+
+    def sample(self, ctxt, publisher_id, event_type, payload, metadata):
+        # TODO(r-mibu): requeue on error
+        self.evaluator.evaluate_events(payload)
+
+
+class EventAlarmEvaluationService(service.Service):
+
+    def __init__(self, conf):
+        super(EventAlarmEvaluationService, self).__init__()
+        self.conf = conf
+        self.storage_conn = storage.get_connection_from_config(self.conf)
+        self.evaluator = event.EventAlarmEvaluator(
+            self.conf,
+            rpc.RPCAlarmNotifier(self.conf))
+
+    def start(self):
+        super(EventAlarmEvaluationService, self).start()
+        self.listener = messaging.get_notification_listener(
+            messaging.get_transport(),
+            oslo_messaging.Target(topic=self.conf.event_alarm_topic),
+            EventAlarmEndpoint(self.evaluator))
+        self.listener.start()
+        # Add a dummy thread to have wait() working
+        self.tg.add_timer(604800, lambda: None)
+
+    def stop(self):
+        self.listener.stop()
+        self.listener.wait()
+        super(EventAlarmEvaluationService, self).stop()

aodh/opts.py

@@ -20,6 +20,7 @@ import aodh.api.controllers.v2.alarms
 import aodh.coordination
 import aodh.evaluator
 import aodh.evaluator.gnocchi
+import aodh.event
 import aodh.notifier.rest
 import aodh.rpc
 import aodh.service
@@ -32,6 +33,7 @@ def list_opts():
          itertools.chain(
              aodh.evaluator.OPTS,
              aodh.evaluator.gnocchi.OPTS,
+             aodh.event.OPTS,
              aodh.notifier.rest.OPTS,
              aodh.service.OPTS,
              aodh.rpc.OPTS,

aodh/tests/test_event.py

@@ -0,0 +1,43 @@
+#
+# Copyright 2015 NEC Corporation.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+import mock
+
+from oslo_config import fixture as fixture_config
+
+from aodh import event
+from aodh import service
+from aodh.tests import base as tests_base
+
+
+class TestEventAlarmEvaluationService(tests_base.BaseTestCase):
+
+    def setUp(self):
+        super(TestEventAlarmEvaluationService, self).setUp()
+
+        conf = service.prepare_service([])
+        self.CONF = self.useFixture(fixture_config.Config(conf)).conf
+        self.storage_conn = mock.MagicMock()
+        self.setup_messaging(self.CONF)
+        with mock.patch('aodh.storage.get_connection_from_config',
+                        return_value=self.storage_conn):
+            self.service = event.EventAlarmEvaluationService(self.CONF)
+
+    def test_start_service(self):
+        listener = mock.Mock()
+        with mock.patch('aodh.messaging.get_notification_listener',
+                        return_value=listener):
+            self.service.start()
+        self.assertTrue(listener.start.called)

setup.cfg

@@ -65,6 +65,7 @@ console_scripts =
     aodh-expirer = aodh.cmd.eventlet.storage:expirer
     aodh-evaluator = aodh.cmd.eventlet.alarm:evaluator
     aodh-notifier = aodh.cmd.eventlet.alarm:notifier
+    aodh-listener = aodh.cmd.eventlet.alarm:listener
 
 oslo.config.opts =
     aodh = aodh.opts:list_opts