Merge lp:~blamar/glance/notifier-strategy into lp:~johannes.erdfelt/glance/notifier

[Brian Lamar]

[Johannes Erdfelt (johannes.erdfelt)]

I like the direction of your changes. The code was a port of the Tempo notification code, which in turn was a port of the nova notification code, but it was a mistake not to take the opportunity and consider if it could be simpler.

review: Approve

=== modified file 'Authors'
--- Authors	2011-07-26 04:24:28 +0000
+++ Authors	2011-07-26 18:44:34 +0000
@@ -1,4 +1,5 @@
 Andrey Brindeyev <[e-mail]>
+Brian Lamar <[e-mail]>
 Brian Waldon <[e-mail]>
 Christopher MacGown <[e-mail]>
 Cory Wright <[e-mail]>
 
=== modified file 'bin/glance-api'
--- bin/glance-api	2011-07-22 17:59:11 +0000
+++ bin/glance-api	2011-07-26 18:44:34 +0000
@@ -36,7 +36,6 @@
 
 from glance import version
 from glance.common import config
-from glance.common import notifier
 from glance.common import wsgi
 
 
@@ -60,8 +59,6 @@
     try:
         conf, app = config.load_paste_app('glance-api', options, args)
 
-        notifier.configure_notifier(conf)
-
         server = wsgi.Server()
         server.start(app, int(conf['bind_port']), conf['bind_host'])
         server.wait()
 
=== modified file 'bin/glance-registry'
--- bin/glance-registry	2011-07-22 17:59:11 +0000
+++ bin/glance-registry	2011-07-26 18:44:34 +0000
@@ -36,7 +36,6 @@
 
 from glance import version
 from glance.common import config
-from glance.common import notifier
 from glance.common import wsgi
 
 
@@ -60,8 +59,6 @@
     try:
         conf, app = config.load_paste_app('glance-registry', options, args)
 
-        notifier.configure_notifier(conf)
-
         server = wsgi.Server()
         server.start(app, int(conf['bind_port']), conf['bind_host'])
         server.wait()
 
=== modified file 'glance/api/v1/images.py'
--- glance/api/v1/images.py	2011-07-26 04:48:33 +0000
+++ glance/api/v1/images.py	2011-07-26 18:44:34 +0000
@@ -73,6 +73,7 @@
 
     def __init__(self, options):
         self.options = options
+        self.notifier = notifier.Notifier(options)
 
     def index(self, req):
         """
@@ -354,7 +355,7 @@
                                            image_id,
                                            {'checksum': checksum,
                                             'size': size})
-            notifier.notify('image.upload', 'INFO', image_meta)
+            self.notifier.info('image.upload', image_meta)
 
             return location
 
@@ -362,14 +363,14 @@
             msg = ("Attempt to upload duplicate image: %s") % str(e)
             logger.error(msg)
             self._safe_kill(req, image_id)
-            notifier.notify('image.upload', 'ERROR', msg)
+            self.notifier.error('image.upload', msg)
             raise HTTPConflict(msg, request=req)
 
         except exception.NotAuthorized, e:
             msg = ("Unauthorized upload attempt: %s") % str(e)
             logger.error(msg)
             self._safe_kill(req, image_id)
-            notifier.notify('image.upload', 'ERROR', msg)
+            self.notifier.error('image.upload', msg)
             raise HTTPForbidden(msg, request=req,
                                 content_type='text/plain')
 
@@ -377,7 +378,7 @@
             msg = ("Error uploading image: %s") % str(e)
             logger.error(msg)
             self._safe_kill(req, image_id)
-            notifier.notify('image.upload', 'ERROR', msg)
+            self.notifier.error('image.upload', msg)
             raise HTTPBadRequest(msg, request=req)
 
     def _activate(self, req, image_id, location):
@@ -524,10 +525,10 @@
                    % locals())
             for line in msg.split('\n'):
                 logger.error(line)
-            notifier.notify('image.update', 'ERROR', msg)
+            self.notifier.error('image.update', msg)
             raise HTTPBadRequest(msg, request=req, content_type="text/plain")
         else:
-            notifier.notify('image.update', 'INFO', image_meta)
+            self.notifier.info('image.update', image_meta)
 
         return {'image_meta': image_meta}
 
@@ -559,7 +560,7 @@
             schedule_delete_from_backend(image['location'], self.options,
                                          req.context, id)
         registry.delete_image_metadata(self.options, req.context, id)
-        notifier.notify('image.delete', 'INFO', id)
+        self.notifier.info('image.delete', id)
 
     def get_store_or_400(self, request, store_name):
         """
 
=== modified file 'glance/common/client.py'
--- glance/common/client.py	2011-07-20 22:53:44 +0000
+++ glance/common/client.py	2011-07-26 18:44:34 +0000
@@ -56,6 +56,12 @@
         self.auth_tok = auth_tok
         self.connection = None
 
+    def set_auth_token(self, auth_tok):
+        """
+        Updates the authentication token for this client connection.
+        """
+        self.auth_tok = auth_tok
+
     def get_connection_type(self):
         """
         Returns the proper connection type
 
=== modified file 'glance/common/exception.py'
--- glance/common/exception.py	2011-07-13 20:15:46 +0000
+++ glance/common/exception.py	2011-07-26 18:44:34 +0000
@@ -145,3 +145,7 @@
 
 class InvalidContentType(GlanceException):
     message = "Invalid content type %(content_type)s"
+
+
+class InvalidNotifierStrategy(GlanceException):
+    message = "'%(strategy)s' is not an available notifier strategy."
 
=== modified file 'glance/common/notifier.py'
--- glance/common/notifier.py	2011-07-26 17:11:27 +0000
+++ glance/common/notifier.py	2011-07-26 18:44:34 +0000
@@ -20,148 +20,128 @@
 import socket
 import uuid
 
-from kombu.connection import BrokerConnection
+import kombu.connection
 
 from glance.common import config
-
-
-WARN = 'WARN'
-INFO = 'INFO'
-ERROR = 'ERROR'
-CRITICAL = 'CRITICAL'
-DEBUG = 'DEBUG'
-
-log_levels = (DEBUG, WARN, INFO, ERROR, CRITICAL)
-
-_DRIVER = None
-
-
-class BadPriorityException(Exception):
-    pass
-
-
-def configure_notifier(options):
-    global _DRIVER
-    notification_driver = config.get_option(options, 'notification_driver',
-            type='str', default='logging')
-    _DRIVER = _get_notifier_driver(notification_driver)(options)
-
-
-def notify(event_type, priority, payload):
-    """
-    Sends a notification using the specified driver
-
-    Notify parameters:
-
-    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': utils.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))
-
-    msg = dict(message_id=str(uuid.uuid4()),
-               publisher_id=socket.gethostname(),
-               event_type=event_type,
-               priority=priority,
-               payload=payload,
-               timestamp=str(datetime.datetime.utcnow()))
-
-    _DRIVER.notify(msg)
-
-
-class Notifier(object):
-    def __init__(self, options):
-        self.level = config.get_option(options, 'default_notification_level',
-                type='str', default='INFO')
-
-    def notify(self, msg):
-        raise NotImplementedError()
-
-
-class NoopNotifier(Notifier):
-    def notify(self, msg):
-        pass
-
-
-class LoggingNotifier(Notifier):
-    def __init__(self, options):
-        super(LoggingNotifier, self).__init__(options)
-        self._setup_logger()
-
-    def _setup_logger(self):
-        str2log_level = {
-            'DEBUG': logging.DEBUG,
-            'INFO': logging.INFO,
-            'WARN': logging.WARN,
-            'ERROR': logging.ERROR,
-            'CRITICAL': logging.CRITICAL}
-        self.level = str2log_level[self.level]
+from glance.common import exception
+
+
+class NoopStrategy(object):
+    """A notifier that does nothing when called."""
+
+    def __init__(self, options):
+        pass
+
+    def warn(self, msg):
+        pass
+
+    def info(self, msg):
+        pass
+
+    def error(self, msg):
+        pass
+
+
+class LoggingStrategy(object):
+    """A notifier that calls logging when called."""
+
+    def __init__(self, options):
         self.logger = logging.getLogger('glance.notifier.logging_notifier')
 
-    def notify(self, msg):
-        self.logger.log(self.level, msg)
-
-
-class RabbitNotifier(Notifier):
+    def warn(self, msg):
+        self.logger.warn(msg)
+
+    def info(self, msg):
+        self.logger.info(msg)
+
+    def error(self, msg):
+        self.logger.error(msg)
+
+
+class RabbitStrategy(object):
+    """A notifier that puts a message on a queue when called."""
+
     def __init__(self, options):
-        super(RabbitNotifier, self).__init__(options)
-        host = config.get_option(options, 'rabbit_host',
-                type='str', default='localhost')
-        port = config.get_option(options, 'rabbit_port',
-                type='int', default=5672)
-        use_ssl = config.get_option(options, 'rabbit_use_ssl',
-                type='bool', default=False)
-        userid = config.get_option(options, 'rabbit_userid',
-                type='str', default='guest')
-        password = config.get_option(options, 'rabbit_password',
-                type='str', default='guest')
-        virtual_host = config.get_option(options, 'rabbit_virtual_host',
-                type='str', default='/')
-
-        self.connection = BrokerConnection(
-                hostname=host,
-                userid=userid,
-                password=password,
-                virtual_host=virtual_host,
-                ssl=use_ssl)
-        self.topic = config.get_option(options, 'rabbit_notification_topic',
-                type='str', default='glance_notifications')
-
-    def notify(self, message):
-        priority = message.get('priority', self.level)
+        """Initialize the rabbit notification strategy."""
+        self._options = options
+        host = self._get_option('rabbit_host', 'str', 'localhost')
+        port = self._get_option('rabbit_port', 'int', 5672)
+        use_ssl = self._get_option('rabbit_use_ssl', 'bool', False)
+        userid = self._get_option('rabbit_userid', 'str', 'guest')
+        password = self._get_option('rabbit_password', 'str', 'guest')
+        virtual_host = self._get_option('rabbit_virtual_host', 'str', '/')
+
+        self.connection = kombu.connection.BrokerConnection(
+            hostname=host,
+            userid=userid,
+            password=password,
+            virtual_host=virtual_host,
+            ssl=use_ssl)
+
+        self.topic = self._get_option('rabbit_notification_topic',
+                                      'str',
+                                      'glance_notifications')
+
+    def _get_option(self, name, datatype, default):
+        """Retrieve a configuration option."""
+        return config.get_option(self._options,
+                                 name,
+                                 type=datatype,
+                                 default=default)
+
+    def _send_message(self, message, priority):
         topic = "%s.%s" % (self.topic, priority)
         queue = self.connection.SimpleQueue(topic)
         queue.put(message, serializer="json")
         queue.close()
 
-
-def _get_notifier_driver(driver):
-    if driver == "logging":
-        return LoggingNotifier
-    elif driver == "rabbit":
-        return RabbitNotifier
-    else:
-        return NoopNotifier
+    def warn(self, msg):
+        self._send_message(msg, "WARN")
+
+    def info(self, msg):
+        self._send_message(msg, "INFO")
+
+    def error(self, msg):
+        self._send_message(msg, "ERROR")
+
+
+class Notifier(object):
+    """Uses a notification strategy to send out messages about events."""
+
+    STRATEGIES = {
+        "logging": LoggingStrategy,
+        "rabbit": RabbitStrategy,
+        "noop": NoopStrategy,
+        "default": NoopStrategy,
+    }
+
+    def __init__(self, options, strategy=None):
+        strategy = config.get_option(options, "notifier_strategy",
+                                     type="str", default="default")
+        try:
+            self.strategy = self.STRATEGIES[strategy](options)
+        except KeyError:
+            raise exception.InvalidNotifierStrategy(strategy=strategy)
+
+    @staticmethod
+    def generate_message(event_type, priority, payload):
+        return {
+            "message_id": str(uuid.uuid4()),
+            "publisher_id": socket.gethostname(),
+            "event_type": event_type,
+            "priority": priority,
+            "payload": payload,
+            "timestamp": str(datetime.datetime.utcnow()),
+        }
+
+    def warn(self, event_type, payload):
+        msg = self.generate_message(event_type, "WARN", payload)
+        self.strategy.warn(msg)
+
+    def info(self, event_type, payload):
+        msg = self.generate_message(event_type, "INFO", payload)
+        self.strategy.info(msg)
+
+    def error(self, event_type, payload):
+        msg = self.generate_message(event_type, "ERROR", payload)
+        self.strategy.error(msg)
 
=== modified file 'tests/functional/test_scrubber.py'
--- tests/functional/test_scrubber.py	2011-07-22 23:08:43 +0000
+++ tests/functional/test_scrubber.py	2011-07-26 18:44:34 +0000
@@ -100,10 +100,26 @@
         for rec in recs:
             self.assertEqual(rec['status'], 'pending_delete')
 
-        # Wait 15 seconds for the scrubber to scrub
-        time.sleep(15)
-
-        recs = list(self.run_sql_cmd(sql))
+        # NOTE(jkoelker) The build servers sometimes take longer than
+        #                15 seconds to scrub. Give it up to 5 min, checking
+        #                checking every 15 seconds. When/if it flips to
+        #                deleted, bail immediatly.
+        deleted = set()
+        recs = []
+        for _ in xrange(20):
+            time.sleep(15)
+
+            recs = list(self.run_sql_cmd(sql))
+            self.assertTrue(recs)
+
+            # NOTE(jkoelker) Reset the deleted set for this loop
+            deleted = set()
+            for rec in recs:
+                deleted.add(rec['status'] == 'deleted')
+
+            if False not in deleted:
+                break
+
         self.assertTrue(recs)
         for rec in recs:
             self.assertEqual(rec['status'], 'deleted')
 
=== modified file 'tests/stubs.py'
--- tests/stubs.py	2011-07-26 04:25:03 +0000
+++ tests/stubs.py	2011-07-26 18:44:34 +0000
@@ -31,7 +31,6 @@
 import glance.common.client
 from glance.common import context
 from glance.common import exception
-import glance.common.notifier
 from glance.registry import server as rserver
 from glance.api import v1 as server
 import glance.store
@@ -476,10 +475,3 @@
               fake_datastore.image_get_all_pending_delete)
     stubs.Set(glance.registry.db.api, 'image_get_all',
               fake_datastore.image_get_all)
-
-
-def stub_out_notifier(stubs):
-    def notify(event_type, priority, payload):
-        pass
-
-    stubs.Set(glance.common.notifier, 'notify', notify)
 
=== modified file 'tests/unit/test_api.py'
--- tests/unit/test_api.py	2011-07-26 04:25:03 +0000
+++ tests/unit/test_api.py	2011-07-26 18:44:34 +0000
@@ -1451,7 +1451,6 @@
         self.stubs = stubout.StubOutForTesting()
         stubs.stub_out_registry_and_store_server(self.stubs)
         stubs.stub_out_registry_db_image_api(self.stubs)
-        stubs.stub_out_notifier(self.stubs)
         stubs.stub_out_filesystem_backend()
         sql_connection = os.environ.get('GLANCE_SQL_CONNECTION', "sqlite://")
         options = {'verbose': VERBOSE,
 
=== modified file 'tests/unit/test_clients.py'
--- tests/unit/test_clients.py	2011-07-26 04:25:03 +0000
+++ tests/unit/test_clients.py	2011-07-26 18:44:34 +0000
@@ -897,7 +897,6 @@
         stubs.stub_out_registry_db_image_api(self.stubs)
         stubs.stub_out_registry_and_store_server(self.stubs)
         stubs.stub_out_filesystem_backend()
-        stubs.stub_out_notifier(self.stubs)
         self.client = client.Client("0.0.0.0", doc_root="")
 
     def tearDown(self):
 
=== modified file 'tests/unit/test_notifier.py'
--- tests/unit/test_notifier.py	2011-07-26 13:58:04 +0000
+++ tests/unit/test_notifier.py	2011-07-26 18:44:34 +0000
@@ -15,93 +15,109 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
-import os
-import stubout
+import logging
 import unittest
 
-from glance import client
 from glance.common import exception
-import glance.common.notifier
-
-from tests import stubs
-
-
-TEST_IMAGE_META = {'name': 'test_image',
-                  'is_public': False,
-                  'disk_format': 'raw',
-                  'container_format': 'ovf'}
-
-
-class TestNotifier(unittest.TestCase):
+from glance.common import notifier
+
+
+class TestInvalidNotifier(unittest.TestCase):
     """Test that notifications are generated appropriately"""
 
-    def setUp(self):
-        """Establish a clean test environment"""
-        self.stubs = stubout.StubOutForTesting()
-        stubs.stub_out_registry_db_image_api(self.stubs)
-        stubs.stub_out_registry_and_store_server(self.stubs)
-        stubs.stub_out_filesystem_backend()
-        self.client = client.Client("0.0.0.0", doc_root="")
-        self.notification = (None, None)
-
-        def notify(event_type, priority, payload):
-            self.notification = (event_type, priority)
-
-        self.stubs.Set(glance.common.notifier, 'notify', notify)
-
-    def tearDown(self):
-        """Clear the test environment"""
-        stubs.clean_out_fake_filesystem_backend()
-        self.stubs.UnsetAll()
-
-    def test_create_notify(self):
-        """Test image create notification."""
-
-        self.client.add_image(TEST_IMAGE_META, 'foobar')
-
-        self.assertEquals(self.notification, ('image.upload', 'INFO'))
-
-    def test_update_notify(self):
-        """Test image update notification. """
-
-        image_meta = self.client.add_image(TEST_IMAGE_META, 'foobar')
-        self.client.update_image(image_meta['id'])
-
-        self.assertEquals(self.notification, ('image.update', 'INFO'))
-
-    def test_delete_notify(self):
-        """Test image delete notification. """
-
-        image_meta = self.client.add_image(TEST_IMAGE_META, 'foobar')
-        self.client.delete_image(image_meta['id'])
-
-        self.assertEquals(self.notification, ('image.delete', 'INFO'))
-
-    def test_create_error_notify(self):
-        """Test image create error notification."""
-
-        def boom(options, context, image_id, image_meta, purge_props=False):
-            if 'checksum' in image_meta:
-                raise Exception('boom')
-            return image_meta
-
-        self.stubs.Set(glance.registry, 'update_image_metadata', boom)
-
-        try:
-            self.client.add_image(TEST_IMAGE_META, 'foobar')
-        except exception.Invalid:
-            pass
-
-        self.assertEquals(self.notification, ('image.upload', 'ERROR'))
-
-    def test_update_error_notify(self):
-        """Test image update error notification."""
-
-        image_meta = self.client.add_image(TEST_IMAGE_META, 'foobar')
-        try:
-            self.client.update_image(image_meta['id'],
-                                     {'container_format': 'invalid'})
-        except exception.Invalid:
-            pass
-
-        self.assertEquals(self.notification, ('image.update', 'ERROR'))
+    def test_cannot_create(self):
+        options = {"notifier_strategy": "invalid_notifier"}
+        self.assertRaises(exception.InvalidNotifierStrategy,
+                          notifier.Notifier,
+                          options)
+
+
+class TestLoggingNotifier(unittest.TestCase):
+    """Test the logging notifier is selected and works properly."""
+
+    def setUp(self):
+        options = {"notifier_strategy": "logging"}
+        self.called = False
+        self.logger = logging.getLogger("glance.notifier.logging_notifier")
+        self.notifier = notifier.Notifier(options)
+
+    def _called(self, msg):
+        self.called = msg
+
+    def test_warn(self):
+        self.logger.warn = self._called
+        self.notifier.warn("test_event", "test_message")
+        if self.called is False:
+            self.fail("Did not call logging library correctly.")
+
+    def test_info(self):
+        self.logger.info = self._called
+        self.notifier.info("test_event", "test_message")
+        if self.called is False:
+            self.fail("Did not call logging library correctly.")
+
+    def test_erorr(self):
+        self.logger.error = self._called
+        self.notifier.error("test_event", "test_message")
+        if self.called is False:
+            self.fail("Did not call logging library correctly.")
+
+
+class TestNoopNotifier(unittest.TestCase):
+    """Test that the noop notifier works...and does nothing?"""
+
+    def setUp(self):
+        options = {"notifier_strategy": "noop"}
+        self.notifier = notifier.Notifier(options)
+
+    def test_warn(self):
+        self.notifier.warn("test_event", "test_message")
+
+    def test_info(self):
+        self.notifier.info("test_event", "test_message")
+
+    def test_error(self):
+        self.notifier.error("test_event", "test_message")
+
+
+class TestRabbitNotifier(unittest.TestCase):
+    """Test AMQP/Rabbit notifier works."""
+
+    def setUp(self):
+        notifier.RabbitStrategy._send_message = self._send_message
+        self.called = False
+        options = {"notifier_strategy": "rabbit"}
+        self.notifier = notifier.Notifier(options)
+
+    def _send_message(self, message, priority):
+        self.called = {
+            "message": message,
+            "priority": priority,
+        }
+
+    def test_warn(self):
+        self.notifier.warn("test_event", "test_message")
+
+        if self.called is False:
+            self.fail("Did not call _send_message properly.")
+
+        self.assertEquals("test_message", self.called["message"]["payload"])
+        self.assertEquals("WARN", self.called["message"]["priority"])
+
+    def test_info(self):
+        self.notifier.info("test_event", "test_message")
+
+        if self.called is False:
+            self.fail("Did not call _send_message properly.")
+
+        self.assertEquals("test_message", self.called["message"]["payload"])
+        self.assertEquals("INFO", self.called["message"]["priority"])
+
+    def test_error(self):
+        self.notifier.error("test_event", "test_message")
+
+        if self.called is False:
+            self.fail("Did not call _send_message properly.")
+
+        self.assertEquals("test_message", self.called["message"]["payload"])
+        self.assertEquals("ERROR", self.called["message"]["priority"])
 
=== modified file 'tools/pip-requires'
--- tools/pip-requires	2011-07-12 09:09:36 +0000
+++ tools/pip-requires	2011-07-26 18:44:34 +0000
@@ -19,3 +19,4 @@
 httplib2
 hashlib
 xattr
+kombu