Refactored using low-level DB query on volume_metadata

Author: leust

cinder/db/api.py

@@ -281,6 +281,12 @@ def volume_get_all(context, marker=None, limit=None, sort_keys=None,
                                offset=offset)
 
 
+def get_host_by_volume_metadata(key, value, filters=None):
+    """Returns the host with the most volumes matching volume metadata."""
+    return IMPL.get_host_by_volume_metadata(key, value,
+                                            filters=filters)
+
+
 def calculate_resource_count(context, resource_type, filters):
     return IMPL.calculate_resource_count(context, resource_type, filters)
 

cinder/db/sqlalchemy/api.py

@@ -2176,6 +2176,38 @@ def volume_get_all(context, marker=None, limit=None, sort_keys=None,
         return query.all()
 
 
+def get_host_by_volume_metadata(meta_key, meta_value, filters=None):
+    session = get_session()
+    count_label = func.count().label("n")
+    query = session.query(
+        func.substring_index(models.Volume.host, '@', 1).label("h"),
+        count_label
+    ).join(
+        models.VolumeMetadata,
+        models.VolumeMetadata.volume_id == models.Volume.id
+    ).filter(
+        models.VolumeMetadata.key == meta_key,
+        models.VolumeMetadata.value == meta_value,
+        models.Volume.deleted == 0,
+        models.Volume.host.isnot(None)
+    )
+
+    if filters:
+        az = filters.get('availability_zone')
+        if az:
+            query = query.filter(
+                models.Volume.availability_zone == az)
+
+    query = query.group_by("h")\
+        .order_by(desc(count_label)).limit(1)
+
+    with session.begin():
+        result = query.first()
+        if result:
+            return result[0]
+    return None
+
+
 @require_context
 def get_volume_summary(context, project_only, filters=None):
     """Retrieves all volumes summary.

cinder/objects/volume.py

@@ -692,16 +692,3 @@ def get_all_active_by_window(cls, context, begin, end):
         expected_attrs = cls._get_expected_attrs(context)
         return base.obj_make_list(context, cls(context), objects.Volume,
                                   volumes, expected_attrs=expected_attrs)
-
-    @classmethod
-    def get_all_by_metadata(cls, context, project_id, metadata, marker=None,
-                            limit=None, sort_keys=None, sort_dirs=None,
-                            filters=None, offset=None):
-        query_filters = {'metadata': metadata}
-        if filters:
-            query_filters.update(filters)
-        volumes = db.volume_get_all_by_project(
-            context, project_id, marker, limit, sort_keys=sort_keys,
-            sort_dirs=sort_dirs, filters=query_filters, offset=offset)
-        return base.obj_make_list(context, cls(context), objects.Volume,
-                                  volumes)

cinder/scheduler/base_filter.py

@@ -92,7 +92,6 @@ def get_filtered_objects(self, filter_classes, objs,
                       each resource.
         """
         list_objs = list(objs)
-        all_objs = list(objs)
         LOG.debug("Starting with %d host(s)", len(list_objs))
         # The 'part_filter_results' list just tracks the number of hosts
         # before and after the filter, unless the filter returns zero
@@ -106,11 +105,6 @@ def get_filtered_objects(self, filter_classes, objs,
             start_count = len(list_objs)
             filter_class = filter_cls()
 
-            # SAP
-            # All available backends are needed in ShardFilter
-            if hasattr(filter_class, 'all_backend_states'):
-                setattr(filter_class, 'all_backend_states', all_objs)
-
             if filter_class.run_filter_for_index(index):
                 objs = filter_class.filter_all(list_objs, filter_properties)
                 if objs is None:

cinder/scheduler/filters/shard_filter.py

@@ -19,8 +19,7 @@
 from oslo_config import cfg
 from oslo_log import log as logging
 
-from cinder import context as cinder_context
-from cinder.objects.volume import VolumeList
+from cinder import db
 from cinder.scheduler import filters
 from cinder.service_auth import SERVICE_USER_GROUP
 from cinder import utils as cinder_utils
@@ -66,9 +65,6 @@ class ShardFilter(filters.BaseBackendFilter):
     _CAPABILITY_NAME = 'vcenter-shard'
     _ALL_SHARDS = "sharding_enabled"
 
-    # To be populated by the host manager
-    all_backend_states = []
-
     def _get_keystone_adapter(self):
         """Return a keystone adapter
 
@@ -151,11 +147,7 @@ def _get_shards(self, project_id):
         return self._PROJECT_SHARD_CACHE.get(project_id)
 
     def _is_vmware(self, backend_state):
-        # We only need the shard filter for vmware based pools
         if backend_state.vendor_name != 'VMware':
-            LOG.info(
-                "Shard Filter ignoring backend %s as it's not "
-                "vmware based driver", backend_state.backend_id)
             return False
         return True
 
@@ -172,8 +164,10 @@ def _filter_by_k8s_cluster(self, backends, filter_properties):
         project_id = vol_props.get('project_id', None)
         metadata = vol_props.get('metadata', {})
 
+        is_vmware = any(self._is_vmware(b) for b in backends)
         if (not metadata or not project_id
-                or spec.get('snapshot_id')):
+                or spec.get('snapshot_id')
+                or not is_vmware):
             return backends
 
         cluster_name = metadata.get(CSI_CLUSTER_METADATA_KEY)
@@ -186,42 +180,28 @@ def _filter_by_k8s_cluster(self, backends, filter_properties):
         if availability_zone:
             query_filters = {'availability_zone': availability_zone}
 
-        query_metadata = {CSI_CLUSTER_METADATA_KEY: cluster_name}
-        k8s_volumes = VolumeList.get_all_by_metadata(
-            cinder_context.get_admin_context(),
-            project_id, query_metadata,
+        k8s_host = db.get_host_by_volume_metadata(
+            key=CSI_CLUSTER_METADATA_KEY,
+            value=cluster_name,
             filters=query_filters)
 
-        if not k8s_volumes:
-            return backends
-
-        k8s_hosts = set(volume_utils.extract_host(v.host, 'host')
-                        for v in k8s_volumes
-                        if v.id != spec.get('volume_id') and v.host)
-        if not k8s_hosts:
+        if not k8s_host:
             return backends
 
-        def _backend_shards(backend_state):
-            cap = backend_state.capabilities.get(self._CAPABILITY_NAME)
-            return cap.split(',') if cap else []
-
-        hosts_shards_map = {
-            volume_utils.extract_host(bs.host, 'host'):
-                _backend_shards(bs)
-            for bs in self.all_backend_states}
-
-        k8s_shards = set()
-        for host in k8s_hosts:
-            shards = hosts_shards_map[host]
-            k8s_shards.update(shards)
-
         return [
             b for b in backends if
             (not self._is_vmware(b) or
-             set(_backend_shards(b)) & k8s_shards)
+             volume_utils.extract_host(b.host, 'host') == k8s_host)
         ]
 
     def _backend_passes(self, backend_state, filter_properties):
+        # We only need the shard filter for vmware based pools
+        if not self._is_vmware(backend_state):
+            LOG.info(
+                "Shard Filter ignoring backend %s as it's not "
+                "vmware based driver", backend_state.backend_id)
+            return True
+
         spec = filter_properties.get('request_spec', {})
         vol = spec.get('volume_properties', {})
         project_id = vol.get('project_id', None)

cinder/tests/unit/objects/test_volume.py

@@ -694,24 +694,3 @@ def test_populate_consistencygroup(self, mock_db_grp_create):
         volume.populate_consistencygroup()
         self.assertEqual(volume.group_id, volume.consistencygroup_id)
         self.assertEqual(volume.group.id, volume.consistencygroup.id)
-
-    @mock.patch('cinder.db.volume_get_all_by_project')
-    def test_get_by_metadata(self, get_all_by_project):
-        db_volume = fake_volume.fake_db_volume()
-        get_all_by_project.return_value = [db_volume]
-
-        volumes = objects.VolumeList.get_all_by_metadata(
-            self.context, mock.sentinel.project_id, mock.sentinel.metadata,
-            mock.sentinel.marker, mock.sentinel.limit,
-            mock.sentinel.sorted_keys, mock.sentinel.sorted_dirs)
-
-        self.assertEqual(1, len(volumes))
-        TestVolume._compare(self, db_volume, volumes[0])
-
-        get_all_by_project.assert_called_once_with(
-            self.context, mock.sentinel.project_id,
-            mock.sentinel.marker, mock.sentinel.limit,
-            sort_keys=mock.sentinel.sorted_keys,
-            sort_dirs=mock.sentinel.sorted_dirs,
-            filters={'metadata': mock.sentinel.metadata},
-            offset=None)

cinder/tests/unit/scheduler/test_base_filter.py

@@ -90,15 +90,6 @@ class FakeFilter5(BaseFakeFilter):
     pass
 
 
-class FakeFilterAllBackends(BaseFakeFilter):
-    """Derives from BaseFakeFilter but has no entry point.
-
-    Should not be included.
-    """
-    all_backend_states = None
-    pass
-
-
 class FilterA(base_filter.BaseFilter):
     def filter_all(self, list_objs, filter_properties):
         # return all but the first object
@@ -181,17 +172,3 @@ def test_get_filtered_objects_with_filter_run_once(self):
             result = self._get_filtered_objects(filter_classes, index=2)
             self.assertEqual(filter_objs_expected, result)
             self.assertEqual(1, fake5_filter_all.call_count)
-
-    @mock.patch.object(FakeFilterAllBackends, 'all_backend_states',
-                       new_callable=mock.PropertyMock)
-    @mock.patch.object(FakeFilterAllBackends, 'filter_all')
-    def test_get_filtered_objects_with_all_backend_states(self, filter_all,
-                                                          all_backends):
-        filter_objs_expected = [1, 2, 3, 4]
-        filter_classes = [FakeFilterAllBackends]
-        filter_all.return_value = filter_objs_expected
-        self._get_filtered_objects(filter_classes)
-        all_backends.assert_has_calls([
-            mock.call(),
-            mock.call(filter_objs_expected)
-        ])

cinder/tests/unit/scheduler/test_shard_filter.py

@@ -16,7 +16,6 @@
 
 from cinder import context
 from cinder.tests.unit import fake_constants
-from cinder.tests.unit import fake_volume
 from cinder.tests.unit.scheduler import fakes
 from cinder.tests.unit.scheduler.test_host_filters \
     import BackendFiltersTestCase
@@ -229,9 +228,10 @@ def test_noop_for_find_backend_by_connector_without_hint(self):
         self.backend_passes(host, self.props)
 
     @mock.patch('cinder.context.get_admin_context')
-    @mock.patch('cinder.objects.volume.VolumeList.get_all_by_metadata')
-    def test_same_shard_for_k8s_volumes(self, mock_get_all,
+    @mock.patch('cinder.db.get_host_by_volume_metadata')
+    def test_same_shard_for_k8s_volumes(self, mock_get_hosts,
                                         mock_get_context):
+        CSI_KEY = 'cinder.csi.openstack.org/cluster'
         all_backends = [
             fakes.FakeBackendState(
                 'volume-vc-a-0@backend#pool1',
@@ -244,13 +244,9 @@ def test_same_shard_for_k8s_volumes(self, mock_get_all,
         ]
         mock_get_context.return_value = self.context
         fake_meta = {
-            'cinder.csi.openstack.org/cluster': 'cluster-1',
+            CSI_KEY: 'cluster-1',
         }
-        mock_get_all.return_value = [
-            fake_volume.fake_volume_obj(self.context, metadata=fake_meta,
-                                        host='volume-vc-a-1@backend#pool3')
-        ]
-        self.filt_cls.all_backend_states = all_backends
+        mock_get_hosts.return_value = 'volume-vc-a-1'
         self.filt_cls._PROJECT_SHARD_CACHE['baz'] = ['sharding_enabled',
                                                      'vc-a-1']
         filter_props = dict(self.props)
@@ -264,8 +260,8 @@ def test_same_shard_for_k8s_volumes(self, mock_get_all,
 
         filtered = self.filt_cls.filter_all(all_backends, filter_props)
 
-        mock_get_all.assert_called_once_with(
-            self.context, 'baz', fake_meta, filters={
+        mock_get_hosts.assert_called_once_with(
+            key=CSI_KEY, value=fake_meta[CSI_KEY], filters={
                 'availability_zone': 'az-1'
             })
         self.assertEqual(len(filtered), 1)