Add notifications for sending an image

An image.send notification is to be sent to the notifier every time an image is
transmitted from glance. This can be used to track things such as bandwidth
usage.

Addresses bug 914440

Change-Id: If8b6504c4250fa6444d17d611de43d9704ca9aae

doc/source/notifications.rst

@@ -17,7 +17,7 @@
 Notifications
 =============
 
-Notifications can be generated for each upload, update or delete image
+Notifications can be generated for each send, upload, update or delete image
 event. These can be used for auditing, troubleshooting, etc.
 
 Strategies
@@ -70,16 +70,28 @@ Every message contains a handful of attributes.
 Payload
 -------
 
-WARN and ERROR events contain a text message in the payload.
+* image.send
+
+  The payload for INFO, WARN, and ERROR events contain the following::
+
+  image_id - ID of the image (UUID)
+  owner_id - Tenant or User ID that owns this image (string)
+  receiver_tenant_id - Tenant ID of the account receiving the image (string)
+  receiver_user_id - User ID of the account receiving the image (string)
+  destination_ip
+  bytes_sent - The number of bytes actually sent
 
 * image.upload
 
   For INFO events, it is the image metadata.
+  WARN and ERROR events contain a text message in the payload.
 
 * image.update
 
   For INFO events, it is the image metadata.
+  WARN and ERROR events contain a text message in the payload.
 
 * image.delete
 
   For INFO events, it is the image id.
+  WARN and ERROR events contain a text message in the payload.

glance/api/middleware/cache.py

@@ -46,7 +46,7 @@ class CacheFilter(wsgi.Middleware):
     def __init__(self, app, conf, **local_conf):
         self.conf = conf
         self.cache = image_cache.ImageCache(conf)
-        self.serializer = images.ImageSerializer()
+        self.serializer = images.ImageSerializer(conf)
         logger.info(_("Initialized image cache middleware"))
         super(CacheFilter, self).__init__(app)
 
@@ -80,7 +80,7 @@ def process_request(self, request):
             try:
                 image_meta = registry.get_image_metadata(context, image_id)
 
-                response = webob.Response()
+                response = webob.Response(request=request)
                 return self.serializer.show(response, {
                     'image_iterator': image_iterator,
                     'image_meta': image_meta})

glance/api/v1/images.py

@@ -631,6 +631,10 @@ def update(self, request):
 class ImageSerializer(wsgi.JSONResponseSerializer):
     """Handles serialization of specific controller method responses."""
 
+    def __init__(self, conf):
+        self.conf = conf
+        self.notifier = notifier.Notifier(conf)
+
     def _inject_location_header(self, response, image_meta):
         location = self._get_image_location(image_meta)
         response.headers['Location'] = location
@@ -666,6 +670,28 @@ def meta(self, response, result):
         self._inject_checksum_header(response, image_meta)
         return response
 
+    def image_send_notification(self, bytes_written, expected_size,
+                                image_meta, request):
+        """Send an image.send message to the notifier."""
+        try:
+            context = request.context
+            payload = {
+                'bytes_sent': bytes_written,
+                'image_id': image_meta['id'],
+                'owner_id': image_meta['owner'],
+                'receiver_tenant_id': context.tenant,
+                'receiver_user_id': context.user,
+                'destination_ip': request.remote_addr,
+            }
+            if bytes_written != expected_size:
+                self.notifier.error('image.send', payload)
+            else:
+                self.notifier.info('image.send', payload)
+        except Exception, err:
+            msg = _("An error occurred during image.send"
+                    " notification: %(err)s") % locals()
+            logger.error(msg)
+
     def show(self, response, result):
         image_meta = result['image_meta']
         image_id = image_meta['id']
@@ -678,6 +704,16 @@ def show(self, response, result):
         # size of the image file. See LP Bug #882585.
         def checked_iter(image_id, expected_size, image_iter):
             bytes_written = 0
+
+            def notify_image_sent_hook(env):
+                self.image_send_notification(bytes_written, expected_size,
+                                             image_meta, response.request)
+
+            # Add hook to process after response is fully sent
+            if 'eventlet.posthooks' in response.environ:
+                response.environ['eventlet.posthooks'].append(
+                    (notify_image_sent_hook, (), {}))
+
             try:
                 for chunk in image_iter:
                     yield chunk
@@ -731,5 +767,5 @@ def create(self, response, result):
 def create_resource(conf):
     """Images resource factory method"""
     deserializer = ImageDeserializer()
-    serializer = ImageSerializer()
+    serializer = ImageSerializer(conf)
     return wsgi.Resource(Controller(conf), deserializer, serializer)

glance/common/wsgi.py

@@ -391,7 +391,7 @@ def __call__(self, request):
         action_result = self.dispatch(self.controller, action,
                                       request, **action_args)
         try:
-            response = webob.Response()
+            response = webob.Response(request=request)
             self.dispatch(self.serializer, action, response, action_result)
             return response
 

glance/tests/unit/test_api.py

@@ -25,6 +25,7 @@
 import stubout
 import webob
 
+from glance.api.v1 import images
 from glance.api.v1 import router
 from glance.common import context
 from glance.common import utils
@@ -2688,3 +2689,149 @@ def test_delete_member(self):
 
         res = req.get_response(self.api)
         self.assertEquals(res.status_int, webob.exc.HTTPUnauthorized.code)
+
+
+class TestImageSerializer(unittest.TestCase):
+    def setUp(self):
+        """Establish a clean test environment"""
+        self.stubs = stubout.StubOutForTesting()
+        stubs.stub_out_registry_and_store_server(self.stubs)
+        stubs.stub_out_filesystem_backend()
+        conf = test_utils.TestConfigOpts(CONF)
+        self.receiving_user = 'fake_user'
+        self.receiving_tenant = 2
+        self.context = rcontext.RequestContext(is_admin=True,
+                                               user=self.receiving_user,
+                                               tenant=self.receiving_tenant)
+        self.serializer = images.ImageSerializer(conf)
+
+        def image_iter():
+            for x in ['chunk', '678911234', '56789']:
+                yield x
+
+        self.FIXTURE = {
+             'image_iterator': image_iter(),
+             'image_meta': {
+                 'id': UUID2,
+                 'name': 'fake image #2',
+                 'status': 'active',
+                 'disk_format': 'vhd',
+                 'container_format': 'ovf',
+                 'is_public': True,
+                 'created_at': datetime.datetime.utcnow(),
+                 'updated_at': datetime.datetime.utcnow(),
+                 'deleted_at': None,
+                 'deleted': False,
+                 'checksum': None,
+                 'size': 19,
+                 'owner': _gen_uuid(),
+                 'location': "file:///tmp/glance-tests/2",
+                 'properties': {}}
+             }
+
+    def tearDown(self):
+        """Clear the test environment"""
+        stubs.clean_out_fake_filesystem_backend()
+        self.stubs.UnsetAll()
+
+    def test_meta(self):
+        exp_headers = {'x-image-meta-id': UUID2,
+                       'x-image-meta-location': 'file:///tmp/glance-tests/2',
+                       'ETag': self.FIXTURE['image_meta']['checksum'],
+                       'x-image-meta-name': 'fake image #2'}
+        req = webob.Request.blank("/images/%s" % UUID2)
+        req.method = 'HEAD'
+        req.remote_addr = "1.2.3.4"
+        req.context = self.context
+        response = webob.Response(request=req)
+        self.serializer.meta(response, self.FIXTURE)
+        for key, value in exp_headers.iteritems():
+            self.assertEquals(value, response.headers[key])
+
+    def test_show(self):
+        exp_headers = {'x-image-meta-id': UUID2,
+                       'x-image-meta-location': 'file:///tmp/glance-tests/2',
+                       'ETag': self.FIXTURE['image_meta']['checksum'],
+                       'x-image-meta-name': 'fake image #2'}
+        req = webob.Request.blank("/images/%s" % UUID2)
+        req.method = 'GET'
+        req.context = self.context
+        response = webob.Response(request=req)
+
+        self.serializer.show(response, self.FIXTURE)
+        for key, value in exp_headers.iteritems():
+            self.assertEquals(value, response.headers[key])
+
+        self.assertEqual(response.body, 'chunk67891123456789')
+
+    def test_show_notify(self):
+        """Make sure an eventlet posthook for notify_image_sent is added."""
+        req = webob.Request.blank("/images/%s" % UUID2)
+        req.method = 'GET'
+        req.context = self.context
+        response = webob.Response(request=req)
+        response.environ['eventlet.posthooks'] = []
+
+        self.serializer.show(response, self.FIXTURE)
+
+        #just make sure the app_iter is called
+        for chunk in response.app_iter:
+            pass
+
+        self.assertNotEqual(response.environ['eventlet.posthooks'], [])
+
+    def test_image_send_notification(self):
+        req = webob.Request.blank("/images/%s" % UUID2)
+        req.method = 'GET'
+        req.remote_addr = '1.2.3.4'
+        req.context = self.context
+
+        image_meta = self.FIXTURE['image_meta']
+        called = {"notified": False}
+        expected_payload = {
+            'bytes_sent': 19,
+            'image_id': UUID2,
+            'owner_id': image_meta['owner'],
+            'receiver_tenant_id': self.receiving_tenant,
+            'receiver_user_id': self.receiving_user,
+            'destination_ip': '1.2.3.4',
+            }
+
+        def fake_info(_event_type, _payload):
+            self.assertDictEqual(_payload, expected_payload)
+            called['notified'] = True
+
+        self.stubs.Set(self.serializer.notifier, 'info', fake_info)
+
+        self.serializer.image_send_notification(19, 19, image_meta, req)
+
+        self.assertTrue(called['notified'])
+
+    def test_image_send_notification_error(self):
+        """Ensure image.send notification is sent on error."""
+        req = webob.Request.blank("/images/%s" % UUID2)
+        req.method = 'GET'
+        req.remote_addr = '1.2.3.4'
+        req.context = self.context
+
+        image_meta = self.FIXTURE['image_meta']
+        called = {"notified": False}
+        expected_payload = {
+            'bytes_sent': 17,
+            'image_id': UUID2,
+            'owner_id': image_meta['owner'],
+            'receiver_tenant_id': self.receiving_tenant,
+            'receiver_user_id': self.receiving_user,
+            'destination_ip': '1.2.3.4',
+            }
+
+        def fake_error(_event_type, _payload):
+            self.assertDictEqual(_payload, expected_payload)
+            called['notified'] = True
+
+        self.stubs.Set(self.serializer.notifier, 'error', fake_error)
+
+        #expected and actually sent bytes differ
+        self.serializer.image_send_notification(17, 19, image_meta, req)
+
+        self.assertTrue(called['notified'])