Make sure stale image metadata is not used

When uploading a volume to an image, Cinder checks to see if there's
any image metadata already present on the volume, and if so, it adds
it to the image-create request so that Glance will persist these
additional properties on the new image.  If during upload preparation,
Cinder wants to add specific metadata to the new image (for example,
a new encryption key id), we need to ensure that the new value for
this property is preferred to a value in the volume_glance_metadata
table.

Closes-bug: #1844725

Change-Id: Iba3c5fa4db87641a84eb22a0fc93294dd55a3132

cinder/tests/unit/image/test_glance.py

@@ -918,9 +918,13 @@ def test_translate_to_glance(self):
             'size': 2,
             'min_disk': 2,
             'min_ram': 2,
+            'cinder_encryption_key_deletion_policy': 'outer',
+            # note that a key duplicated in the 'properties' dict
+            # will overwrite the "outer" value
             'properties': {'kernel_id': 'foo',
                            'ramdisk_id': 'bar',
-                           'x_billinginfo': '123'},
+                           'x_billinginfo': '123',
+                           'cinder_encryption_key_deletion_policy': 'NOPE'},
         }
 
         actual = service._translate_to_glance(metadata)
@@ -929,6 +933,36 @@ def test_translate_to_glance(self):
             'size': 2,
             'min_disk': 2,
             'min_ram': 2,
+            'cinder_encryption_key_deletion_policy': 'NOPE',
+            'kernel_id': 'foo',
+            'ramdisk_id': 'bar',
+            'x_billinginfo': '123',
+        }
+        self.assertEqual(expected, actual)
+
+    def test_translate_to_glance_no_properties_element(self):
+        """Show _translate does not remove arbitrary flat properties"""
+        client = glance_stubs.StubGlanceClient()
+        service = self._create_image_service(client)
+
+        metadata = {
+            'id': 1,
+            'cinder_encryption_key_deletion_policy': 'baz',
+            'size': 2,
+            'min_disk': 2,
+            'min_ram': 2,
+            'kernel_id': 'foo',
+            'ramdisk_id': 'bar',
+            'x_billinginfo': '123',
+        }
+
+        actual = service._translate_to_glance(metadata)
+        expected = {
+            'id': 1,
+            'cinder_encryption_key_deletion_policy': 'baz',
+            'size': 2,
+            'min_disk': 2,
+            'min_ram': 2,
             'kernel_id': 'foo',
             'ramdisk_id': 'bar',
             'x_billinginfo': '123',

cinder/tests/unit/volume/test_image.py

@@ -744,3 +744,37 @@ def test_copy_volume_to_image_maintenance(self):
                           volume,
                           test_meta1,
                           force=True)
+
+
+class CopyVolumeToImagePrivateFunctionsTestCase(cinder.test.TestCase):
+
+    @mock.patch('cinder.volume.api.API.get_volume_image_metadata',
+                return_value={'some_key': 'some_value',
+                              'cinder_encryption_key_id': 'stale_value'})
+    def test_merge_volume_image_meta(self, mock_get_img_meta):
+        # this is what we're passing to copy_volume_to_image
+        image_meta = {
+            'container_format': 'bare',
+            'disk_format': 'raw',
+            'cinder_encryption_key_id': 'correct_value'
+        }
+        self.assertNotIn('properties', image_meta)
+
+        volume_api = cinder.volume.api.API()
+        volume_api._merge_volume_image_meta(None, None, image_meta)
+        # we've got 'properties' now
+        self.assertIn('properties', image_meta)
+        # verify the key_id is what we expect
+        self.assertEqual(image_meta['cinder_encryption_key_id'],
+                         'correct_value')
+
+        translate = cinder.image.glance.GlanceImageService._translate_to_glance
+        sent_to_glance = translate(image_meta)
+
+        # this is correct, glance gets a "flat" dict of properties
+        self.assertNotIn('properties', sent_to_glance)
+
+        # make sure the image would be created in Glance with the
+        # correct key_id
+        self.assertEqual(image_meta['cinder_encryption_key_id'],
+                         sent_to_glance['cinder_encryption_key_id'])

cinder/volume/api.py

@@ -1276,6 +1276,29 @@ def get_list_volumes_image_metadata(self, context, volume_id_list):
                                                      meta_entry['value']})
         return results
 
+    def _merge_volume_image_meta(self, context, volume, image_meta):
+        """Merges the image metadata from volume into image_meta"""
+        glance_core_props = CONF.glance_core_properties
+        if glance_core_props:
+            try:
+                vol_img_metadata = self.get_volume_image_metadata(
+                    context, volume)
+                custom_property_set = (
+                    set(vol_img_metadata).difference(glance_core_props))
+                if custom_property_set:
+                    # only include elements that haven't already been
+                    # assigned values
+                    filtered_property_set = custom_property_set.difference(
+                        image_meta)
+                    image_meta['properties'] = {
+                        custom_prop: vol_img_metadata[custom_prop]
+                        for custom_prop in filtered_property_set}
+            except exception.GlanceMetadataNotFound:
+                # If volume is not created from image, No glance metadata
+                # would be available for that volume in
+                # volume glance metadata table
+                pass
+
     def copy_volume_to_image(self, context, volume, metadata, force):
         """Create a new image from the specified volume."""
         if not CONF.enable_force_upload and force:
@@ -1298,23 +1321,7 @@ def copy_volume_to_image(self, context, volume, metadata, force):
             raise exception.InvalidVolume(reason=msg)
 
         try:
-            glance_core_props = CONF.glance_core_properties
-            if glance_core_props:
-                try:
-                    vol_img_metadata = self.get_volume_image_metadata(
-                        context, volume)
-                    custom_property_set = (
-                        set(vol_img_metadata).difference(glance_core_props))
-                    if custom_property_set:
-                        metadata['properties'] = {
-                            custom_prop: vol_img_metadata[custom_prop]
-                            for custom_prop in custom_property_set}
-                except exception.GlanceMetadataNotFound:
-                    # If volume is not created from image, No glance metadata
-                    # would be available for that volume in
-                    # volume glance metadata table
-                    pass
-
+            self._merge_volume_image_meta(context, volume, metadata)
             recv_metadata = self.image_service.create(context, metadata)
         except Exception:
             # NOTE(geguileo): To mimic behavior before conditional_update we