rbd driver in cinder does not manage glance images multi-location

Currently, rbd driver in cinder does not manage properly glance images with
multiple locations when creating a volume from image.

With multi-location feature in glance, a image can have multiple rbd URIs
associated, even in different ceph clusters. When rbd driver tries to create a
volume as a CoW clone from the image, it only check the direct_url URI instead
of all available locations for the image.

This patch iterates over both direct_url and locations URIs looking for a
clonable one.

Change-Id: I2d50b8af7f8f772ffa10d968616309d81e6940ec
Closes-Bug: 1454638

cinder/tests/unit/test_rbd.py

@@ -1116,7 +1116,7 @@ class ManagedRBDTestCase(test_volume.DriverTestCase):
         driver = self.volume.driver
 
         with mock.patch.object(driver, '_is_cloneable', lambda *args: False):
-            image_loc = (mock.Mock(), mock.Mock())
+            image_loc = (mock.Mock(), None)
             actual = driver.clone_image(mock.Mock(),
                                         mock.Mock(),
                                         image_loc,
@@ -1152,3 +1152,68 @@ class ManagedRBDTestCase(test_volume.DriverTestCase):
                     mock_clone.assert_called_once_with(volume,
                                                        'fi', 'fo', 'fum')
                     mock_resize.assert_called_once_with(volume)
+
+    def test_clone_multilocation_success(self):
+        expected = ({'provider_location': None}, True)
+        driver = self.volume.driver
+
+        def cloneable_side_effect(url_location, image_meta):
+            return url_location == 'rbd://fee/fi/fo/fum'
+
+        with mock.patch.object(self.volume.driver, '_is_cloneable') \
+            as mock_is_cloneable, \
+            mock.patch.object(self.volume.driver, '_clone') as mock_clone, \
+            mock.patch.object(self.volume.driver, '_resize') \
+                as mock_resize:
+                mock_is_cloneable.side_effect = cloneable_side_effect
+                image_loc = ('rbd://bee/bi/bo/bum',
+                             [{'url': 'rbd://bee/bi/bo/bum'},
+                              {'url': 'rbd://fee/fi/fo/fum'}])
+                volume = {'name': 'vol1'}
+                image_meta = mock.sentinel.image_meta
+                image_service = mock.sentinel.image_service
+
+                actual = driver.clone_image(self.context,
+                                            volume,
+                                            image_loc,
+                                            image_meta,
+                                            image_service)
+
+                self.assertEqual(expected, actual)
+                self.assertEqual(2, mock_is_cloneable.call_count)
+                mock_clone.assert_called_once_with(volume,
+                                                   'fi', 'fo', 'fum')
+                mock_is_cloneable.assert_called_with('rbd://fee/fi/fo/fum',
+                                                     image_meta)
+                mock_resize.assert_called_once_with(volume)
+
+    def test_clone_multilocation_failure(self):
+        expected = ({}, False)
+        driver = self.volume.driver
+
+        with mock.patch.object(driver, '_is_cloneable', return_value=False) \
+            as mock_is_cloneable, \
+            mock.patch.object(self.volume.driver, '_clone') as mock_clone, \
+            mock.patch.object(self.volume.driver, '_resize') \
+                as mock_resize:
+                image_loc = ('rbd://bee/bi/bo/bum',
+                             [{'url': 'rbd://bee/bi/bo/bum'},
+                              {'url': 'rbd://fee/fi/fo/fum'}])
+
+                volume = {'name': 'vol1'}
+                image_meta = mock.sentinel.image_meta
+                image_service = mock.sentinel.image_service
+                actual = driver.clone_image(self.context,
+                                            volume,
+                                            image_loc,
+                                            image_meta,
+                                            image_service)
+
+                self.assertEqual(expected, actual)
+                self.assertEqual(2, mock_is_cloneable.call_count)
+                mock_is_cloneable.assert_any_call('rbd://bee/bi/bo/bum',
+                                                  image_meta)
+                mock_is_cloneable.assert_any_call('rbd://fee/fi/fo/fum',
+                                                  image_meta)
+                self.assertFalse(mock_clone.called)
+                self.assertFalse(mock_resize.called)

cinder/volume/drivers/rbd.py

@@ -841,14 +841,27 @@ class RBDDriver(driver.RetypeVD, driver.TransferVD, driver.ExtendVD,
     def clone_image(self, context, volume,
                     image_location, image_meta,
                     image_service):
-        image_location = image_location[0] if image_location else None
-        if image_location is None or not self._is_cloneable(
-                image_location, image_meta):
-            return ({}, False)
-        _prefix, pool, image, snapshot = self._parse_location(image_location)
-        self._clone(volume, pool, image, snapshot)
-        self._resize(volume)
-        return {'provider_location': None}, True
+        if image_location:
+            # Note: image_location[0] is glance image direct_url.
+            # image_location[1] contains the list of all locations (including
+            # direct_url) or None if show_multiple_locations is False in
+            # glance configuration.
+            if image_location[1]:
+                url_locations = [location['url'] for
+                                 location in image_location[1]]
+            else:
+                url_locations = [image_location[0]]
+
+            # iterate all locations to look for a cloneable one.
+            for url_location in url_locations:
+                if url_location and self._is_cloneable(
+                        url_location, image_meta):
+                    _prefix, pool, image, snapshot = \
+                        self._parse_location(url_location)
+                    self._clone(volume, pool, image, snapshot)
+                    self._resize(volume)
+                    return {'provider_location': None}, True
+        return ({}, False)
 
     def _image_conversion_dir(self):
         tmpdir = (self.configuration.volume_tmp_dir or