Adding notifications for volumes

Added notifications for volumes have been added with tests.
This includes create/delete/exists events for volumes.

blueprint nova-notifications

Change-Id: I21b74974fac22c3621ccf7564dc5c0d339f8751a

6 files changed, 236 insertions, 1 deletions

diff --git a/nova/db/api.py b/nova/db/api.py
index ca1e420d7..fed92072d 100644
--- a/nova/db/api.py
+++ b/nova/db/api.py
@@ -1612,6 +1612,13 @@ def volume_type_destroy(context, name):
     return IMPL.volume_type_destroy(context, name)
 
 
+def volume_get_active_by_window(context, begin, end=None, project_id=None):
+    """Get all the volumes inside the window.
+
+    Specifying a project_id will filter for a certain project."""
+    return IMPL.volume_get_active_by_window(context, begin, end, project_id)
+
+
 ####################
 
 
diff --git a/nova/db/sqlalchemy/api.py b/nova/db/sqlalchemy/api.py
index 56ce054d3..4e8e69313 100644
--- a/nova/db/sqlalchemy/api.py
+++ b/nova/db/sqlalchemy/api.py
@@ -4109,6 +4109,23 @@ def volume_type_destroy(context, name):
                         'updated_at': literal_column('updated_at')})
 
 
+@require_context
+def volume_get_active_by_window(context, begin, end=None,
+                                         project_id=None):
+    """Return volumes that were active during window."""
+    session = get_session()
+    query = session.query(models.Volume)
+
+    query = query.filter(or_(models.Volume.deleted_at == None,
+                             models.Volume.deleted_at > begin))
+    if end:
+        query = query.filter(models.Volume.created_at < end)
+    if project_id:
+        query = query.filter_by(project_id=project_id)
+
+    return query.all()
+
+
 ####################
 
 
diff --git a/nova/tests/test_volume.py b/nova/tests/test_volume.py
index d979c3d0e..88e8146b6 100644
--- a/nova/tests/test_volume.py
+++ b/nova/tests/test_volume.py
@@ -29,6 +29,7 @@ from nova import exception
 from nova import db
 from nova import flags
 from nova import log as logging
+from nova.notifier import test_notifier
 from nova.openstack.common import importutils
 import nova.policy
 from nova import rpc
@@ -46,18 +47,21 @@ class VolumeTestCase(test.TestCase):
         super(VolumeTestCase, self).setUp()
         self.compute = importutils.import_object(FLAGS.compute_manager)
         self.flags(connection_type='fake')
+        self.stubs.Set(nova.flags.FLAGS, 'notification_driver',
+                'nova.notifier.test_notifier')
         self.volume = importutils.import_object(FLAGS.volume_manager)
         self.context = context.get_admin_context()
         instance = db.instance_create(self.context, {})
         self.instance_id = instance['id']
         self.instance_uuid = instance['uuid']
+        test_notifier.NOTIFICATIONS = []
 
     def tearDown(self):
         db.instance_destroy(self.context, self.instance_id)
         super(VolumeTestCase, self).tearDown()
 
     @staticmethod
-    def _create_volume(size='0', snapshot_id=None):
+    def _create_volume(size=0, snapshot_id=None):
         """Create a volume object."""
         vol = {}
         vol['size'] = size
@@ -88,11 +92,14 @@ class VolumeTestCase(test.TestCase):
         """Test volume can be created and deleted."""
         volume = self._create_volume()
         volume_id = volume['id']
+        self.assertEquals(len(test_notifier.NOTIFICATIONS), 0)
         self.volume.create_volume(self.context, volume_id)
+        self.assertEquals(len(test_notifier.NOTIFICATIONS), 2)
         self.assertEqual(volume_id, db.volume_get(context.get_admin_context(),
                          volume_id).id)
 
         self.volume.delete_volume(self.context, volume_id)
+        self.assertEquals(len(test_notifier.NOTIFICATIONS), 4)
         self.assertRaises(exception.NotFound,
                           db.volume_get,
                           self.context,
@@ -363,6 +370,30 @@ class VolumeTestCase(test.TestCase):
         self.volume.delete_snapshot(self.context, snapshot_id)
         self.volume.delete_volume(self.context, volume_id)
 
+    def test_create_volume_usage_notification(self):
+        """Ensure create volume generates appropriate usage notification"""
+        volume = self._create_volume()
+        volume_id = volume['id']
+        self.assertEquals(len(test_notifier.NOTIFICATIONS), 0)
+        self.volume.create_volume(self.context, volume_id)
+        self.assertEquals(len(test_notifier.NOTIFICATIONS), 2)
+        msg = test_notifier.NOTIFICATIONS[0]
+        self.assertEquals(msg['event_type'], 'volume.create.start')
+        msg = test_notifier.NOTIFICATIONS[1]
+        self.assertEquals(msg['priority'], 'INFO')
+        self.assertEquals(msg['event_type'], 'volume.create.end')
+        payload = msg['payload']
+        self.assertEquals(payload['tenant_id'], volume['project_id'])
+        self.assertEquals(payload['user_id'], volume['user_id'])
+        self.assertEquals(payload['volume_id'], volume['id'])
+        self.assertEquals(payload['status'], 'creating')
+        self.assertEquals(payload['size'], volume['size'])
+        self.assertTrue('display_name' in payload)
+        self.assertTrue('snapshot_id' in payload)
+        self.assertTrue('launched_at' in payload)
+        self.assertTrue('created_at' in payload)
+        self.volume.delete_volume(self.context, volume_id)
+
 
 class DriverTestCase(test.TestCase):
     """Base Test class for Drivers."""
diff --git a/nova/tests/test_volume_utils.py b/nova/tests/test_volume_utils.py
new file mode 100644
index 000000000..453c0d925
--- /dev/null
+++ b/nova/tests/test_volume_utils.py
@@ -0,0 +1,86 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# 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.
+
+"""Tests For miscellaneous util methods used with volume."""
+
+from nova import db
+from nova import flags
+from nova import context
+from nova import test
+from nova import log as logging
+import nova.image.fake
+from nova.volume import utils as volume_utils
+from nova.notifier import test_notifier
+from nova.openstack.common import importutils
+
+
+LOG = logging.getLogger(__name__)
+FLAGS = flags.FLAGS
+
+
+class UsageInfoTestCase(test.TestCase):
+
+    def setUp(self):
+        super(UsageInfoTestCase, self).setUp()
+        self.flags(connection_type='fake',
+                   stub_network=True,
+                   host='fake')
+        self.stubs.Set(nova.flags.FLAGS, 'notification_driver',
+                'nova.notifier.test_notifier')
+        self.volume = importutils.import_object(FLAGS.volume_manager)
+        self.user_id = 'fake'
+        self.project_id = 'fake'
+        self.snapshot_id = 'fake'
+        self.volume_size = 0
+        self.context = context.RequestContext(self.user_id, self.project_id)
+        test_notifier.NOTIFICATIONS = []
+
+    def _create_volume(self, params={}):
+        """Create a test volume"""
+        vol = {}
+        vol['snapshot_id'] = self.snapshot_id
+        vol['user_id'] = self.user_id
+        vol['project_id'] = self.project_id
+        vol['host'] = FLAGS.host
+        vol['availability_zone'] = FLAGS.storage_availability_zone
+        vol['status'] = "creating"
+        vol['attach_status'] = "detached"
+        vol['size'] = self.volume_size
+        vol.update(params)
+        return db.volume_create(self.context, vol)['id']
+
+    def test_notify_usage_exists(self):
+        """Ensure 'exists' notification generates appropriate usage data."""
+        volume_id = self._create_volume()
+        volume = db.volume_get(self.context, volume_id)
+        volume_utils.notify_usage_exists(self.context, volume)
+        self.assertEquals(len(test_notifier.NOTIFICATIONS), 1)
+        msg = test_notifier.NOTIFICATIONS[0]
+        self.assertEquals(msg['priority'], 'INFO')
+        self.assertEquals(msg['event_type'], 'volume.exists')
+        payload = msg['payload']
+        self.assertEquals(payload['tenant_id'], self.project_id)
+        self.assertEquals(payload['user_id'], self.user_id)
+        self.assertEquals(payload['snapshot_id'], self.snapshot_id)
+        self.assertEquals(payload['volume_id'], volume.id)
+        self.assertEquals(payload['size'], self.volume_size)
+        for attr in ('display_name', 'created_at', 'launched_at',
+                     'status', 'audit_period_beginning',
+                     'audit_period_ending'):
+            self.assertTrue(attr in payload,
+                            msg="Key %s not in payload" % attr)
+        db.volume_destroy(context.get_admin_context(), volume['id'])
diff --git a/nova/volume/manager.py b/nova/volume/manager.py
index 828451f7c..a471ea372 100644
--- a/nova/volume/manager.py
+++ b/nova/volume/manager.py
@@ -49,6 +49,7 @@ from nova.openstack.common import importutils
 from nova import rpc
 from nova import utils
 from nova.volume import volume_types
+from nova.volume import utils as volume_utils
 
 
 LOG = logging.getLogger(__name__)
@@ -106,6 +107,7 @@ class VolumeManager(manager.SchedulerDependentManager):
         """Creates and exports the volume."""
         context = context.elevated()
         volume_ref = self.db.volume_get(context, volume_id)
+        self._notify_about_volume_usage(context, volume_ref, "create.start")
         LOG.info(_("volume %s: creating"), volume_ref['name'])
 
         self.db.volume_update(context,
@@ -145,6 +147,7 @@ class VolumeManager(manager.SchedulerDependentManager):
                                                  'launched_at': now})
         LOG.debug(_("volume %s: created successfully"), volume_ref['name'])
         self._reset_stats()
+        self._notify_about_volume_usage(context, volume_ref, "create.end")
         return volume_id
 
     def delete_volume(self, context, volume_id):
@@ -157,6 +160,7 @@ class VolumeManager(manager.SchedulerDependentManager):
             msg = _("Volume is not local to this node")
             raise exception.NovaException(msg)
 
+        self._notify_about_volume_usage(context, volume_ref, "delete.start")
         self._reset_stats()
         try:
             LOG.debug(_("volume %s: removing export"), volume_ref['name'])
@@ -177,6 +181,7 @@ class VolumeManager(manager.SchedulerDependentManager):
 
         self.db.volume_destroy(context, volume_id)
         LOG.debug(_("volume %s: deleted successfully"), volume_ref['name'])
+        self._notify_about_volume_usage(context, volume_ref, "delete.end")
         return True
 
     def create_snapshot(self, context, volume_id, snapshot_id):
@@ -337,3 +342,9 @@ class VolumeManager(manager.SchedulerDependentManager):
     def notification(self, context, event):
         LOG.info(_("Notification {%s} received"), event)
         self._reset_stats()
+
+    def _notify_about_volume_usage(self, context, volume, event_suffix,
+                                     extra_usage_info=None):
+        volume_utils.notify_about_volume_usage(
+                context, volume, event_suffix,
+                extra_usage_info=extra_usage_info, host=self.host)
diff --git a/nova/volume/utils.py b/nova/volume/utils.py
new file mode 100644
index 000000000..94f719bf3
--- /dev/null
+++ b/nova/volume/utils.py
@@ -0,0 +1,83 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright (c) 2012 OpenStack, LLC.
+#
+#    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.
+
+"""Volume-related Utilities and helpers."""
+
+from nova import flags
+from nova import log as logging
+from nova import utils
+from nova.notifier import api as notifier_api
+
+
+FLAGS = flags.FLAGS
+LOG = logging.getLogger(__name__)
+
+
+def notify_usage_exists(context, volume_ref, current_period=False):
+    """ Generates 'exists' notification for a volume for usage auditing
+        purposes.
+
+        Generates usage for last completed period, unless 'current_period'
+        is True."""
+    begin, end = utils.last_completed_audit_period()
+    if current_period:
+        audit_start = end
+        audit_end = utils.utcnow()
+    else:
+        audit_start = begin
+        audit_end = end
+
+    extra_usage_info = dict(audit_period_beginning=str(audit_start),
+                            audit_period_ending=str(audit_end))
+
+    notify_about_volume_usage(
+            context, volume_ref, 'exists', extra_usage_info=extra_usage_info)
+
+
+def _usage_from_volume(context, volume_ref, **kw):
+    def null_safe_str(s):
+        return str(s) if s else ''
+
+    usage_info = dict(
+          tenant_id=volume_ref['project_id'],
+          user_id=volume_ref['user_id'],
+          volume_id=volume_ref['id'],
+          volume_type=volume_ref['volume_type'],
+          display_name=volume_ref['display_name'],
+          launched_at=null_safe_str(volume_ref['launched_at']),
+          created_at=null_safe_str(volume_ref['created_at']),
+          status=volume_ref['status'],
+          snapshot_id=volume_ref['snapshot_id'],
+          size=volume_ref['size'])
+
+    usage_info.update(kw)
+    return usage_info
+
+
+def notify_about_volume_usage(context, volume, event_suffix,
+                                extra_usage_info=None, host=None):
+    if not host:
+        host = FLAGS.host
+
+    if not extra_usage_info:
+        extra_usage_info = {}
+
+    usage_info = _usage_from_volume(
+            context, volume, **extra_usage_info)
+
+    notifier_api.notify(context, 'volume.%s' % host,
+                        'volume.%s' % event_suffix,
+                        notifier_api.INFO, usage_info)