Merge pull request #175 from sapcc/k8s_sharding

K8S volumes to be scheduled in the same shard

Author: hemna

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/scheduler/filters/shard_filter.py

@@ -19,9 +19,11 @@
 from oslo_config import cfg
 from oslo_log import log as logging
 
+from cinder import db
 from cinder.scheduler import filters
 from cinder.service_auth import SERVICE_USER_GROUP
 from cinder import utils as cinder_utils
+from cinder.volume import volume_utils
 
 
 LOG = logging.getLogger(__name__)
@@ -40,6 +42,8 @@
                  service_type='identity')
 CONF.register_opts(keystone_opts, group=KEYSTONE_GROUP)
 
+CSI_CLUSTER_METADATA_KEY = 'cinder.csi.openstack.org/cluster'
+
 
 class ShardFilter(filters.BaseBackendFilter):
     """Filters backends by shard of the project
@@ -142,12 +146,60 @@ def _get_shards(self, project_id):
 
         return self._PROJECT_SHARD_CACHE.get(project_id)
 
-    def backend_passes(self, backend_state, filter_properties):
-        # We only need the shard filter for vmware based pools
+    def _is_vmware(self, backend_state):
         if backend_state.vendor_name != 'VMware':
+            return False
+        return True
+
+    def filter_all(self, filter_obj_list, filter_properties):
+        backends = self._filter_by_k8s_cluster(filter_obj_list,
+                                               filter_properties)
+
+        return [b for b in backends if
+                self._backend_passes(b, filter_properties)]
+
+    def _filter_by_k8s_cluster(self, backends, filter_properties):
+        spec = filter_properties.get('request_spec', {})
+        vol_props = spec.get('volume_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 not is_vmware):
+            return backends
+
+        cluster_name = metadata.get(CSI_CLUSTER_METADATA_KEY)
+        if not cluster_name:
+            return backends
+
+        props = spec.get('resource_properties', {})
+        availability_zone = props.get('availability_zone')
+        query_filters = None
+        if availability_zone:
+            query_filters = {'availability_zone': availability_zone}
+
+        k8s_host = db.get_host_by_volume_metadata(
+            key=CSI_CLUSTER_METADATA_KEY,
+            value=cluster_name,
+            filters=query_filters)
+
+        if not k8s_host:
+            return backends
+
+        return [
+            b for b in backends if
+            (not self._is_vmware(b) or
+             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)
+                "Shard Filter ignoring backend %s as it's not "
+                "vmware based driver", backend_state.backend_id)
             return True
 
         spec = filter_properties.get('request_spec', {})

cinder/tests/unit/scheduler/test_shard_filter.py

@@ -14,6 +14,8 @@
 import time
 from unittest import mock
 
+from cinder import context
+from cinder.tests.unit import fake_constants
 from cinder.tests.unit.scheduler import fakes
 from cinder.tests.unit.scheduler.test_host_filters \
     import BackendFiltersTestCase
@@ -37,6 +39,8 @@ def setUp(self):
                 }
             }
         }
+        self.context = context.RequestContext(fake_constants.USER_ID,
+                                              fake_constants.PROJECT_ID)
 
     @mock.patch('cinder.scheduler.filters.shard_filter.'
                 'ShardFilter._update_cache')
@@ -77,7 +81,7 @@ def test_shard_project_not_found(self, mock_update_cache):
         host = fakes.FakeBackendState('host1',
                                       {'capabilities': caps,
                                        'vendor_name': VMWARE_VENDOR})
-        self.assertFalse(self.filt_cls.backend_passes(host, self.props))
+        self.backend_no_pass(host, self.props)
 
     def test_snapshot(self):
         snap_props = {
@@ -90,7 +94,7 @@ def test_snapshot(self):
         host = fakes.FakeBackendState('host1',
                                       {'capabilities': caps,
                                        'vendor_name': VMWARE_VENDOR})
-        self.assertTrue(self.filt_cls.backend_passes(host, snap_props))
+        self.backend_passes(host, snap_props)
 
     def test_snapshot_None(self):
         snap_props = {
@@ -103,57 +107,57 @@ def test_snapshot_None(self):
         host = fakes.FakeBackendState('host1',
                                       {'capabilities': caps,
                                        'vendor_name': VMWARE_VENDOR})
-        self.assertFalse(self.filt_cls.backend_passes(host, snap_props))
+        self.backend_no_pass(host, snap_props)
 
     def test_shard_project_no_shards(self):
         caps = {'vcenter-shard': 'vc-a-1'}
         self.filt_cls._PROJECT_SHARD_CACHE['foo'] = []
         host = fakes.FakeBackendState('host1',
                                       {'capabilities': caps,
                                        'vendor_name': VMWARE_VENDOR})
-        self.assertFalse(self.filt_cls.backend_passes(host, self.props))
+        self.backend_no_pass(host, self.props)
 
     def test_backend_without_shard(self):
         host = fakes.FakeBackendState('host1', {'vendor_name': VMWARE_VENDOR})
-        self.assertFalse(self.filt_cls.backend_passes(host, self.props))
+        self.backend_no_pass(host, self.props)
 
     def test_backend_shards_dont_match(self):
         caps = {'vcenter-shard': 'vc-a-1'}
         host = fakes.FakeBackendState('host1',
                                       {'capabilities': caps,
                                        'vendor_name': VMWARE_VENDOR})
-        self.assertFalse(self.filt_cls.backend_passes(host, self.props))
+        self.backend_no_pass(host, self.props)
 
     def test_backend_shards_match(self):
         caps = {'vcenter-shard': 'vc-b-0'}
         host = fakes.FakeBackendState('host1',
                                       {'capabilities': caps,
                                        'vendor_name': VMWARE_VENDOR})
-        self.assertTrue(self.filt_cls.backend_passes(host, self.props))
+        self.backend_passes(host, self.props)
 
     def test_shard_override_matches(self):
         caps = {'vcenter-shard': 'vc-a-1'}
         host = fakes.FakeBackendState('host1',
                                       {'capabilities': caps,
                                        'vendor_name': VMWARE_VENDOR})
         self.props['scheduler_hints'] = {'vcenter-shard': 'vc-a-1'}
-        self.assertTrue(self.filt_cls.backend_passes(host, self.props))
+        self.backend_passes(host, self.props)
 
     def test_shard_override_no_match(self):
         caps = {'vcenter-shard': 'vc-a-0'}
         host = fakes.FakeBackendState('host1',
                                       {'capabilities': caps,
                                        'vendor_name': VMWARE_VENDOR})
         self.props['scheduler_hints'] = {'vcenter-shard': 'vc-a-1'}
-        self.assertFalse(self.filt_cls.backend_passes(host, self.props))
+        self.backend_no_pass(host, self.props)
 
     def test_shard_override_no_data(self):
         caps = {'vcenter-shard': 'vc-a-0'}
         host = fakes.FakeBackendState('host1',
                                       {'capabilities': caps,
                                        'vendor_name': VMWARE_VENDOR})
         self.props['scheduler_hints'] = {'vcenter-shard': None}
-        self.assertFalse(self.filt_cls.backend_passes(host, self.props))
+        self.backend_no_pass(host, self.props)
 
     def test_sharding_enabled_any_backend_match(self):
         self.filt_cls._PROJECT_SHARD_CACHE['baz'] = ['sharding_enabled']
@@ -162,7 +166,7 @@ def test_sharding_enabled_any_backend_match(self):
         host = fakes.FakeBackendState('host1',
                                       {'capabilities': caps,
                                        'vendor_name': VMWARE_VENDOR})
-        self.assertTrue(self.filt_cls.backend_passes(host, self.props))
+        self.backend_passes(host, self.props)
 
     def test_sharding_enabled_and_single_shard_any_backend_match(self):
         self.filt_cls._PROJECT_SHARD_CACHE['baz'] = ['sharding_enabled',
@@ -172,7 +176,7 @@ def test_sharding_enabled_and_single_shard_any_backend_match(self):
         host = fakes.FakeBackendState('host1',
                                       {'capabilities': caps,
                                        'vendor_name': VMWARE_VENDOR})
-        self.assertTrue(self.filt_cls.backend_passes(host, self.props))
+        self.backend_passes(host, self.props)
 
     def test_scheduler_hints_override_sharding_enabled(self):
         self.filt_cls._PROJECT_SHARD_CACHE['baz'] = ['sharding_enabled']
@@ -182,12 +186,12 @@ def test_scheduler_hints_override_sharding_enabled(self):
         host = fakes.FakeBackendState('host0',
                                       {'capabilities': caps0,
                                        'vendor_name': VMWARE_VENDOR})
-        self.assertFalse(self.filt_cls.backend_passes(host, self.props))
+        self.backend_no_pass(host, self.props)
         caps1 = {'vcenter-shard': 'vc-a-1'}
         host = fakes.FakeBackendState('host1',
                                       {'capabilities': caps1,
                                        'vendor_name': VMWARE_VENDOR})
-        self.assertTrue(self.filt_cls.backend_passes(host, self.props))
+        self.backend_passes(host, self.props)
 
     def test_noop_for_find_backend_by_connector_with_hint(self):
         """Check if we pass any backend
@@ -204,7 +208,7 @@ def test_noop_for_find_backend_by_connector_with_hint(self):
                                        'vendor_name': VMWARE_VENDOR})
         self.props['scheduler_hints'] = {'vcenter-shard': 'vc-a-1'}
         self.props['request_spec']['operation'] = 'find_backend_for_connector'
-        self.assertTrue(self.filt_cls.backend_passes(host, self.props))
+        self.backend_passes(host, self.props)
 
     def test_noop_for_find_backend_by_connector_without_hint(self):
         """Check if we pass any backend
@@ -221,4 +225,52 @@ def test_noop_for_find_backend_by_connector_without_hint(self):
                                       {'capabilities': caps,
                                        'vendor_name': VMWARE_VENDOR})
         self.props['request_spec']['operation'] = 'find_backend_for_connector'
-        self.assertTrue(self.filt_cls.backend_passes(host, self.props))
+        self.backend_passes(host, self.props)
+
+    @mock.patch('cinder.context.get_admin_context')
+    @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',
+                {'capabilities': {'vcenter-shard': 'vc-a-0'},
+                 'vendor_name': VMWARE_VENDOR}),
+            fakes.FakeBackendState(
+                'volume-vc-a-1@backend#pool2',
+                {'capabilities': {'vcenter-shard': 'vc-a-1'},
+                 'vendor_name': VMWARE_VENDOR}),
+        ]
+        mock_get_context.return_value = self.context
+        fake_meta = {
+            CSI_KEY: 'cluster-1',
+        }
+        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)
+        filter_props['request_spec']['volume_properties'].update({
+            'project_id': 'baz',
+            'metadata': fake_meta
+        })
+        filter_props['request_spec']['resource_properties'] = {
+            'availability_zone': 'az-1'
+        }
+
+        filtered = self.filt_cls.filter_all(all_backends, filter_props)
+
+        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)
+        self.assertEqual('volume-vc-a-1@backend#pool2', filtered[0].host)
+
+    def backend_passes(self, backend, filter_properties):
+        filtered = self.filt_cls.filter_all([backend], filter_properties)
+        self.assertEqual(backend, filtered[0])
+
+    def backend_no_pass(self, backend, filter_properties):
+        filtered = self.filt_cls.filter_all([backend], filter_properties)
+        self.assertEqual(0, len(filtered))