Implementation of the alarm RPCAlarmNotifier

This change implement the RPCAlarmNotifier to allow
the threshold-evaluator to communicate with the notifier.

Change-Id: I1dd3b5aabe5221a37903a4e87de64702aec94ad2

ceilometer/alarm/rpc.py

@@ -0,0 +1,47 @@
+# -*- encoding: utf-8 -*-
+#
+# Copyright © 2013 eNovance <licensing@enovance.com>
+#
+# Authors: Mehdi Abaakouk <mehdi.abaakouk@enovance.com>
+#
+# 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
+
+from ceilometer.openstack.common import context
+from ceilometer.openstack.common.rpc import proxy as rpc_proxy
+from ceilometer.storage.models import Alarm
+
+OPTS = [
+    cfg.StrOpt('notifier_rpc_topic',
+               default='alarm_notifier',
+               help='the topic ceilometer uses for alarm notifier messages'),
+]
+
+cfg.CONF.register_opts(OPTS, group='alarm')
+
+
+class RPCAlarmNotifier(rpc_proxy.RpcProxy):
+    def __init__(self):
+        super(RPCAlarmNotifier, self).__init__(
+            default_version='1.0',
+            topic=cfg.CONF.alarm.notifier_rpc_topic)
+
+    def notify(self, alarm, state, reason):
+        actions = getattr(alarm, Alarm.ALARM_ACTIONS_MAP[alarm.state])
+        msg = self.make_msg('notify_alarm', data={
+            'actions': actions,
+            'alarm': alarm.alarm_id,
+            'state': state,
+            'reason': reason})
+        self.cast(context.get_admin_context(), msg)

ceilometer/alarm/service.py

@@ -21,12 +21,14 @@
 from oslo.config import cfg
 from stevedore import extension
 
+from ceilometer.alarm import rpc as rpc_alarm
 from ceilometer.service import prepare_service
 from ceilometer.openstack.common import log
 from ceilometer.openstack.common import network_utils
 from ceilometer.openstack.common import service as os_service
 from ceilometer.openstack.common.gettextutils import _
 from ceilometer.openstack.common.rpc import service as rpc_service
+from ceilometer.openstack.common.rpc import dispatcher as rpc_dispatcher
 from ceilometerclient import client as ceiloclient
 
 
@@ -39,6 +41,8 @@ OPTS = [
 ]
 
 cfg.CONF.register_opts(OPTS, group='alarm')
+cfg.CONF.import_opt('notifier_rpc_topic', 'ceilometer.alarm.rpc',
+                    group='alarm')
 
 LOG = log.getLogger(__name__)
 
@@ -52,6 +56,7 @@ class SingletonAlarmService(os_service.Service):
         self.extension_manager = extension.ExtensionManager(
             namespace=self.ALARM_NAMESPACE,
             invoke_on_load=True,
+            invoke_args=(rpc_alarm.RPCAlarmNotifier(),)
         )
 
     def start(self):
@@ -114,6 +119,14 @@ class AlarmNotifierService(rpc_service.Service):
         # Add a dummy thread to have wait() working
         self.tg.add_timer(604800, lambda: None)
 
+    def initialize_service_hook(self, service):
+        LOG.debug('initialize_service_hooks')
+        self.conn.create_worker(
+            cfg.CONF.alarm.notifier_rpc_topic,
+            rpc_dispatcher.RpcDispatcher([self]),
+            'ceilometer.alarm.' + cfg.CONF.alarm.notifier_rpc_topic,
+        )
+
     def _handle_action(self, action, alarm, state, reason):
         try:
             action = network_utils.urlsplit(action)

ceilometer/storage/models.py

@@ -232,6 +232,13 @@ class Alarm(Model):
     ALARM_INSUFFICIENT_DATA = 'insufficient data'
     ALARM_OK = 'ok'
     ALARM_ALARM = 'alarm'
+
+    ALARM_ACTIONS_MAP = {
+        ALARM_INSUFFICIENT_DATA: 'insufficient_data_actions',
+        ALARM_OK: 'ok_actions',
+        ALARM_ALARM: 'alarm_actions',
+    }
+
     """
     An alarm to monitor.
 

etc/ceilometer/ceilometer.conf.sample

@@ -573,6 +573,15 @@
 #rest_notifier_certificate_key=
 
 
+#
+# Options defined in ceilometer.alarm.rpc
+#
+
+# the topic ceilometer uses for alarm notifier messages
+# (string value)
+#notifier_rpc_topic=alarm_notifier
+
+
 #
 # Options defined in ceilometer.alarm.service
 #
@@ -700,4 +709,4 @@
 #password=<None>
 
 
-# Total option count: 132
+# Total option count: 133

tests/alarm/test_notifier.py

@@ -32,6 +32,14 @@ class TestAlarmNotifier(base.TestCase):
         super(TestAlarmNotifier, self).setUp()
         self.service = service.AlarmNotifierService('somehost', 'sometopic')
 
+    @mock.patch('ceilometer.pipeline.setup_pipeline', mock.MagicMock())
+    def test_init_host(self):
+        # If we try to create a real RPC connection, init_host() never
+        # returns. Mock it out so we can establish the service
+        # configuration.
+        with mock.patch('ceilometer.openstack.common.rpc.create_connection'):
+            self.service.start()
+
     def test_notify_alarm(self):
         data = {
             'actions': ['test://'],

tests/alarm/test_rpc.py

@@ -0,0 +1,90 @@
+# -*- encoding: utf-8 -*-
+#
+# Copyright © 2013 eNovance <licensing@enovance.com>
+#
+# Authors: Mehdi Abaakouk <mehdi.abaakouk@enovance.com>
+#
+# 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 uuid
+
+from oslo.config import cfg
+
+from ceilometer.alarm import rpc as rpc_alarm
+from ceilometer.openstack.common import rpc
+from ceilometer.storage.models import Alarm as AlarmModel
+from ceilometer.tests import base
+from ceilometerclient.v2.alarms import Alarm as AlarmClient
+
+
+class TestRPCAlarmNotifier(base.TestCase):
+    def faux_cast(self, context, topic, msg):
+        self.notified.append((topic, msg))
+
+    def setUp(self):
+        super(TestRPCAlarmNotifier, self).setUp()
+        self.notified = []
+        self.stubs.Set(rpc, 'cast', self.faux_cast)
+        self.notifier = rpc_alarm.RPCAlarmNotifier()
+        self.alarms = [
+            AlarmClient(None, info={
+                'name': 'instance_running_hot',
+                'counter_name': 'cpu_util',
+                'comparison_operator': 'gt',
+                'threshold': 80.0,
+                'evaluation_periods': 5,
+                'statistic': 'avg',
+                'state': 'ok',
+                'ok_actions': ['http://host:8080/path'],
+                'user_id': 'foobar',
+                'project_id': 'snafu',
+                'period': 60,
+                'alarm_id': str(uuid.uuid4()),
+                'matching_metadata':{'resource_id':
+                                     'my_instance'}
+            }),
+            AlarmClient(None, info={
+                'name': 'group_running_idle',
+                'counter_name': 'cpu_util',
+                'comparison_operator': 'le',
+                'threshold': 10.0,
+                'statistic': 'max',
+                'evaluation_periods': 4,
+                'state': 'insufficient data',
+                'insufficient_data_actions': ['http://other_host/path'],
+                'user_id': 'foobar',
+                'project_id': 'snafu',
+                'period': 300,
+                'alarm_id': str(uuid.uuid4()),
+                'matching_metadata':{'metadata.user_metadata.AS':
+                                     'my_group'}
+            }),
+        ]
+
+    def test_notify_alarm(self):
+        for i, a in enumerate(self.alarms):
+            self.notifier.notify(a, "ok", "what? %d" % i)
+        self.assertEqual(len(self.notified), 2)
+        for i, a in enumerate(self.alarms):
+            actions = getattr(a, AlarmModel.ALARM_ACTIONS_MAP[a.state])
+            self.assertEqual(self.notified[i][0],
+                             cfg.CONF.alarm.notifier_rpc_topic)
+            self.assertEqual(self.notified[i][1]["args"]["data"]["alarm"],
+                             self.alarms[i].alarm_id)
+            self.assertEqual(self.notified[i][1]["args"]["data"]["actions"],
+                             actions)
+            self.assertEqual(self.notified[i][1]["args"]["data"]["state"],
+                             "ok")
+            self.assertEqual(self.notified[i][1]["args"]["data"]["reason"],
+                             "what? %d" % i)