NFS snapshots

This patch adds support for snapshots to the NFS driver.

This functionality is only enabled if
"nfs_snapshot_support" is set to True in cinder.conf
and nas_secure_file_operations is disabled.

This is because compute nodes running libvirt <1.2.7 will
encounter problems with snapshot deletion, which includes
Ubuntu 14.04 and requires write access to the volumes
created by a different user. Therefore, deployers must
opt-in to this functionality if their environment is known
to support it.

Due libvirt limitations[1], apparmor must be disabled in the
compute nodes or the volume export location be added to
libvirt's default profile template[3].

Clonning volumes is only supported if the source volume
is not attached.

[1] https://bugzilla.redhat.com/show_bug.cgi?id=1361592
[2] http://paste.openstack.org/show/570296/
[3] http://paste.openstack.org/show/570297/

Co-Authored-By: Ankit Agrawal <ankit11.agrawal@nttdata.com>
Co-Authored-By: Jay S. Bryant <jsbryant@us.ibm.com>
Co-Authored-By: Erlon R. Cruz <erlon.cruz@fit-tecnologia.org.br>
Implements: blueprint nfs-snapshots

DocImpact: A new parameter is added and the clonning
limitation must be documented.

Closes-bug: #1614249
Change-Id: Iae35c722eb4b6b7d02a95690abbc07a63da77ce7

cinder/image/image_utils.py

@@ -145,7 +145,8 @@ def _convert_image(prefix, source, dest, out_format, run_as_root=True):
     if duration < 1:
         duration = 1
     try:
-        image_size = qemu_img_info(source, run_as_root=True).virtual_size
+        image_size = qemu_img_info(source,
+                                   run_as_root=run_as_root).virtual_size
     except ValueError as e:
         msg = _LI("The image was successfully converted, but image size "
                   "is unavailable. src %(src)s, dest %(dest)s. %(error)s")

cinder/tests/unit/volume/drivers/test_coho.py

@@ -25,6 +25,7 @@
 from cinder import context
 from cinder import exception
 from cinder import test
+from cinder.tests.unit import fake_volume
 from cinder.volume import configuration as conf
 from cinder.volume.drivers import coho
 from cinder.volume.drivers import nfs
@@ -42,7 +43,7 @@
     'volume_id': 'bcc48c61-9691-4e5f-897c-793686093190',
     'size': 128,
     'volume_type': 'silver',
-    'volume_type_id': 'type-id',
+    'volume_type_id': 'deadbeef-aaaa-bbbb-cccc-deadbeefbeef',
     'metadata': [{'key': 'type',
                   'service_label': 'silver'}],
     'provider_location': 'coho-datastream-addr:/test/path',
@@ -72,7 +73,7 @@
     'updated_at': None,
     'extra_specs': {},
     'deleted_at': None,
-    'id': 'type-id'
+    'id': 'deadbeef-aaaa-bbbb-cccc-deadbeefbeef'
 }
 
 QOS_SPEC = {
@@ -161,6 +162,11 @@ def test_setup_failure_when_coho_rpc_port_is_invalid(self):
     def test_create_volume_with_qos(self):
         drv = coho.CohoDriver(configuration=self.configuration)
 
+        volume = fake_volume.fake_volume_obj(self.context,
+                                             **{'volume_type_id':
+                                                VOLUME['volume_type_id'],
+                                                'provider_location':
+                                                VOLUME['provider_location']})
         mock_remotefs_create = self.mock_object(remotefs.RemoteFSDriver,
                                                 'create_volume')
         mock_rpc_client = self.mock_object(coho, 'CohoRPCClient')
@@ -172,18 +178,18 @@ def test_create_volume_with_qos(self):
         mock_get_admin_context = self.mock_object(context, 'get_admin_context')
         mock_get_admin_context.return_value = 'test'
 
-        drv.create_volume(VOLUME)
+        drv.create_volume(volume)
 
         self.assertTrue(mock_remotefs_create.called)
         self.assertTrue(mock_get_admin_context.called)
-        mock_remotefs_create.assert_has_calls([mock.call(VOLUME)])
+        mock_remotefs_create.assert_has_calls([mock.call(volume)])
         mock_get_volume_type.assert_has_calls(
-            [mock.call('test', VOLUME_TYPE['id'])])
+            [mock.call('test', volume.volume_type_id)])
         mock_get_qos_specs.assert_has_calls(
             [mock.call('test', QOS_SPEC['id'])])
         mock_rpc_client.assert_has_calls(
             [mock.call(ADDR, self.configuration.coho_rpc_port),
-             mock.call().set_qos_policy(os.path.join(PATH, VOLUME['name']),
+             mock.call().set_qos_policy(os.path.join(PATH, volume.name),
                                         QOS)])
 
     def test_create_snapshot(self):