Implements a basic mechanism for pushing notifications out to interested parties. The rationale for implementing notifications this way is that the responsibility for them shouldn't fall to Nova. As such, we simply will be pushing messages to a queue where another worker entirely can be written to push messages around to subscribers.

nova/flags.py

@@ -369,6 +369,9 @@ DEFINE_string('host', socket.gethostname(),
 DEFINE_string('node_availability_zone', 'nova',
               'availability zone of this node')
 
+DEFINE_string('notification_driver',
+              'nova.notifier.no_op_notifier',
+              'Default driver for sending notifications')
 DEFINE_list('memcached_servers', None,
             'Memcached servers or None for in process cache.')
 

nova/notifier/__init__.py

@@ -0,0 +1,14 @@
+# Copyright 2011 OpenStack LLC.
+# All Rights Reserved.
+#
+#    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.

nova/notifier/api.py

@@ -0,0 +1,83 @@
+# Copyright 2011 OpenStack LLC.
+# All Rights Reserved.
+#
+#    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 datetime
+
+import datetime
+import uuid
+
+from nova import flags
+from nova import utils
+
+
+FLAGS = flags.FLAGS
+
+flags.DEFINE_string('default_notification_level', 'INFO',
+                    'Default notification level for outgoing notifications')
+
+WARN = 'WARN'
+INFO = 'INFO'
+ERROR = 'ERROR'
+CRITICAL = 'CRITICAL'
+DEBUG = 'DEBUG'
+
+log_levels = (DEBUG, WARN, INFO, ERROR, CRITICAL)
+
+
+class BadPriorityException(Exception):
+    pass
+
+
+def notify(publisher_id, event_type, priority, payload):
+    """
+    Sends a notification using the specified driver
+
+    Notify parameters:
+
+    publisher_id - the source worker_type.host of the message
+    event_type - the literal type of event (ex. Instance Creation)
+    priority - patterned after the enumeration of Python logging levels in
+               the set (DEBUG, WARN, INFO, ERROR, CRITICAL)
+    payload - A python dictionary of attributes
+
+    Outgoing message format includes the above parameters, and appends the
+    following:
+
+    message_id - a UUID representing the id for this notification
+    timestamp - the GMT timestamp the notification was sent at
+
+    The composite message will be constructed as a dictionary of the above
+    attributes, which will then be sent via the transport mechanism defined
+    by the driver.
+
+    Message example:
+
+    {'message_id': str(uuid.uuid4()),
+     'publisher_id': 'compute.host1',
+     'timestamp': datetime.datetime.utcnow(),
+     'priority': 'WARN',
+     'event_type': 'compute.create_instance',
+     'payload': {'instance_id': 12, ... }}
+
+    """
+    if priority not in log_levels:
+        raise BadPriorityException(
+                 _('%s not in valid priorities' % priority))
+    driver = utils.import_object(FLAGS.notification_driver)
+    msg = dict(message_id=str(uuid.uuid4()),
+                   publisher_id=publisher_id,
+                   event_type=event_type,
+                   priority=priority,
+                   payload=payload,
+                   timestamp=str(datetime.datetime.utcnow()))
+    driver.notify(msg)

nova/notifier/log_notifier.py

@@ -0,0 +1,34 @@
+# Copyright 2011 OpenStack LLC.
+# All Rights Reserved.
+#
+#    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 json
+
+from nova import flags
+from nova import log as logging
+
+
+FLAGS = flags.FLAGS
+
+
+def notify(message):
+    """Notifies the recipient of the desired event given the model.
+    Log notifications using nova's default logging system"""
+
+    priority = message.get('priority',
+                           FLAGS.default_notification_level)
+    priority = priority.lower()
+    logger = logging.getLogger(
+            'nova.notification.%s' % message['event_type'])
+    getattr(logger, priority)(json.dumps(message))

nova/notifier/no_op_notifier.py

@@ -0,0 +1,19 @@
+# Copyright 2011 OpenStack LLC.
+# All Rights Reserved.
+#
+#    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.
+
+
+def notify(message):
+    """Notifies the recipient of the desired event given the model"""
+    pass

nova/notifier/rabbit_notifier.py

@@ -0,0 +1,36 @@
+# Copyright 2011 OpenStack LLC.
+# All Rights Reserved.
+#
+#    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 nova.context
+
+from nova import flags
+from nova import rpc
+
+
+FLAGS = flags.FLAGS
+
+flags.DEFINE_string('notification_topic', 'notifications',
+                    'RabbitMQ topic used for Nova notifications')
+
+
+def notify(message):
+    """Sends a notification to the RabbitMQ"""
+    context = nova.context.get_admin_context()
+    priority = message.get('priority',
+                           FLAGS.default_notification_level)
+    priority = priority.lower()
+    topic = '%s.%s' % (FLAGS.notification_topic, priority)
+    rpc.cast(context, topic, message)

nova/tests/test_notifier.py

@@ -0,0 +1,117 @@
+# Copyright 2011 OpenStack LLC.
+# All Rights Reserved.
+#
+#    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 nova
+
+from nova import context
+from nova import flags
+from nova import rpc
+import nova.notifier.api
+from nova.notifier.api import notify
+from nova.notifier import no_op_notifier
+from nova.notifier import rabbit_notifier
+from nova import test
+
+import stubout
+
+
+class NotifierTestCase(test.TestCase):
+    """Test case for notifications"""
+    def setUp(self):
+        super(NotifierTestCase, self).setUp()
+        self.stubs = stubout.StubOutForTesting()
+
+    def tearDown(self):
+        self.stubs.UnsetAll()
+        super(NotifierTestCase, self).tearDown()
+
+    def test_send_notification(self):
+        self.notify_called = False
+
+        def mock_notify(cls, *args):
+            self.notify_called = True
+
+        self.stubs.Set(nova.notifier.no_op_notifier, 'notify',
+                mock_notify)
+
+        class Mock(object):
+            pass
+        notify('publisher_id', 'event_type',
+                nova.notifier.api.WARN, dict(a=3))
+        self.assertEqual(self.notify_called, True)
+
+    def test_verify_message_format(self):
+        """A test to ensure changing the message format is prohibitively
+        annoying"""
+
+        def message_assert(message):
+            fields = [('publisher_id', 'publisher_id'),
+                      ('event_type', 'event_type'),
+                      ('priority', 'WARN'),
+                      ('payload', dict(a=3))]
+            for k, v in fields:
+                self.assertEqual(message[k], v)
+            self.assertTrue(len(message['message_id']) > 0)
+            self.assertTrue(len(message['timestamp']) > 0)
+
+        self.stubs.Set(nova.notifier.no_op_notifier, 'notify',
+                message_assert)
+        notify('publisher_id', 'event_type',
+                nova.notifier.api.WARN, dict(a=3))
+
+    def test_send_rabbit_notification(self):
+        self.stubs.Set(nova.flags.FLAGS, 'notification_driver',
+                'nova.notifier.rabbit_notifier')
+        self.mock_cast = False
+
+        def mock_cast(cls, *args):
+            self.mock_cast = True
+
+        class Mock(object):
+            pass
+
+        self.stubs.Set(nova.rpc, 'cast', mock_cast)
+        notify('publisher_id', 'event_type',
+                nova.notifier.api.WARN, dict(a=3))
+
+        self.assertEqual(self.mock_cast, True)
+
+    def test_invalid_priority(self):
+        def mock_cast(cls, *args):
+            pass
+
+        class Mock(object):
+            pass
+
+        self.stubs.Set(nova.rpc, 'cast', mock_cast)
+        self.assertRaises(nova.notifier.api.BadPriorityException,
+                notify, 'publisher_id',
+                'event_type', 'not a priority', dict(a=3))
+
+    def test_rabbit_priority_queue(self):
+        self.stubs.Set(nova.flags.FLAGS, 'notification_driver',
+                'nova.notifier.rabbit_notifier')
+        self.stubs.Set(nova.flags.FLAGS, 'notification_topic',
+                'testnotify')
+
+        self.test_topic = None
+
+        def mock_cast(context, topic, msg):
+            self.test_topic = topic
+
+        self.stubs.Set(nova.rpc, 'cast', mock_cast)
+        notify('publisher_id',
+                'event_type', 'DEBUG', dict(a=3))
+        self.assertEqual(self.test_topic, 'testnotify.debug')