WIP: Shard affinity for k8s workloads

nova/scheduler/filters/shard_filter.py

@@ -19,6 +19,9 @@
 from oslo_log import log as logging
 
 import nova.conf
+from nova import context as nova_context
+from nova.objects.instance import Instance
+from nova.objects.instance import InstanceList
 from nova.scheduler import filters
 from nova.scheduler import utils
 from nova import utils as nova_utils
@@ -28,6 +31,9 @@
 CONF = nova.conf.CONF
 
 _SERVICE_AUTH = None
+GARDENER_PREFIX = "kubernetes.io-cluster-shoot--garden--"
+KKS_PREFIX = "kubernikus:kluster"
+HANA_PREFIX = "hana_"
 
 
 class ShardFilter(filters.BaseHostFilter):
@@ -114,11 +120,61 @@ def _get_shards(self, project_id):
 
         return self._PROJECT_SHARD_CACHE.get(project_id)
 
-    def host_passes(self, host_state, spec_obj):
+    def _get_k8s_cluster_instances(self, spec_obj):
+        """If the instance will be part of a K8S cluster, it returns
+        the list of all other instances that are already part of it,
+        if any.
+        """
+        k8s_filter = self._k8s_instance_query_filter(spec_obj)
+
+        if not k8s_filter:
+            return []
+
+        k8s_filter['project_id'] = spec_obj.project_id
+
+        return InstanceList.get_by_filters(
+            nova_context.get_admin_context(), filters=k8s_filter,
+            expected_attrs=['flavor', 'metadata', 'tags'])
+
+    def _k8s_instance_query_filter(self, spec_obj):
+        elevated = nova_context.get_admin_context()
+        instance = Instance.get_by_uuid(elevated, spec_obj.instance_uuid,
+                                        expected_attrs=['tags', 'metadata'])
+
+        # Kubernikus
+        kks_tag = next((t.tag for t in instance.tags
+                        if t.tag.startswith(KKS_PREFIX)), None)
+        if kks_tag:
+            return {'tags': [kks_tag]}
+
+        # Gardener
+        gardener_meta = {k: v for k, v in instance.metadata.items()
+                         if k.starswith(GARDENER_PREFIX)}
+        if gardener_meta:
+            return {'metadata': gardener_meta}
+
+        return None
+
+    def filter_all(self, filter_obj_list, spec_obj):
+        """Yield objects that pass the filter.
+
+        Can be overridden in a subclass, if you need to base filtering
+        decisions on all objects.  Otherwise, one can just override
+        _filter_one() to filter a single object.
+        """
         # Only VMware
         if utils.is_non_vmware_spec(spec_obj):
-            return True
+            for obj in filter_obj_list:
+                yield obj
+            return
+
+        k8s_instances = self._get_k8s_cluster_instances(spec_obj)
+
+        for obj in filter_obj_list:
+            if self._host_passes(obj, spec_obj, k8s_instances):
+                yield obj
 
+    def _host_passes(self, host_state, spec_obj, k8s_instances):
         host_shard_aggrs = [aggr for aggr in host_state.aggregates
                             if aggr.name.startswith(self._SHARD_PREFIX)]
 
@@ -148,18 +204,58 @@ def host_passes(self, host_state, spec_obj):
         if self._ALL_SHARDS in shards:
             LOG.debug('project enabled for all shards %(project_shards)s.',
                       {'project_shards': shards})
-            return True
         elif host_shard_names & set(shards):
             LOG.debug('%(host_state)s shard %(host_shard)s found in project '
                       'shards %(project_shards)s.',
                       {'host_state': host_state,
                        'host_shard': host_shard_names,
                        'project_shards': shards})
-            return True
         else:
             LOG.debug('%(host_state)s shard %(host_shard)s not found in '
                       'project shards %(project_shards)s.',
                       {'host_state': host_state,
                        'host_shard': host_shard_names,
                        'project_shards': shards})
             return False
+
+        if not utils.request_is_resize(spec_obj):
+            # K8S orchestrators are only creating or deleting nodes,
+            # therefore we shouldn't infer with resize/migrate requests.
+            return self._host_passes_k8s(host_state, host_shard_names,
+                                         spec_obj, k8s_instances)
+
+        return True
+
+    def _host_passes_k8s(self, host_state, host_shard_names, spec_obj,
+                         k8s_instances):
+        """Instances of a K8S cluster must end up on the same shard.
+        The K8S cluster is identified by the metadata or tags set
+        by the orchestrator (Gardener or Kubernikus).
+        """
+        if not k8s_instances:
+            # There are no instances in the cluster, yet.
+            # We allow any shard for the first instance.
+            return True
+
+        def _is_hana(flavor):
+            return flavor.name.startswith(HANA_PREFIX)
+
+        def _is_same_category(instance, flavor):
+            """Check whether instance is from the flavor's family."""
+            if _is_hana(flavor):
+                return _is_hana(instance.flavor)
+            return True
+
+        def _instance_matches(instance):
+            if spec_obj.availability_zone:
+                if (instance.availability_zone !=
+                        spec_obj.availability_zone):
+                    return False
+            return _is_same_category(instance, spec_obj.flavor)
+
+        k8s_hosts = set([i.host for i in k8s_instances
+                         if _instance_matches(i)])
+
+        return any(agg.name in host_shard_names and
+                   set(agg.hosts) & k8s_hosts
+                   for agg in host_state.all_aggregates)

nova/scheduler/host_manager.py

@@ -333,7 +333,7 @@ def update_from_host_manager(self, all_aggregates):
         def _locked_update(self, aggregates):
             self.all_aggregates = aggregates
 
-        _locked_update(all_aggregates)
+        _locked_update(self, all_aggregates)
 
 
 class HostManager(object):