feat(scheduler): use /scale endpoints for RC and Deployments to only update replicas during scale events (#1029)

Prior to this a full resource update was being done during a scale event, this could lead to potential pod template updates that were not desired

Fixes #1012

Author: helgi

rootfs/scheduler/mock.py

@@ -12,7 +12,8 @@
 from zlib import adler32
 
 import scheduler
-from scheduler import KubeHTTPClient, KubeHTTPException
+from scheduler import KubeHTTPClient
+from scheduler.exceptions import KubeHTTPException
 
 from django.conf import settings
 from django.core.cache import cache
@@ -79,7 +80,7 @@ def _acquire(self):
 resources = [
     'namespaces', 'nodes', 'pods', 'replicationcontrollers',
     'secrets', 'services', 'events', 'deployments', 'replicasets',
-    'horizontalpodautoscalers'
+    'horizontalpodautoscalers', 'scale',
 ]
 
 
@@ -103,15 +104,48 @@ def cache_key(key):
     return key
 
 
-def get_type(key, pos=-1):
+def get_type(key):
     key = key.strip('/').split('/')
-    if key[pos] in resources:
-        return key[pos]
+
+    # check if is subresource (last element)
+    if len(key) and key[-1] in resources:
+        return key[-1]
+
+    # go back to second last
+    if len(key) > 1 and key[-2] in resources:
+        return key[-2]
 
     # bad if it gets there
     return 'unknown'
 
 
+def get_namespace(url, resource_type):
+    """Inspects the URL and gets namespace from it if there is any"""
+    # correct back to proper namespace API
+    url = url.replace('apis_autoscaling_v1', 'api_v1')
+    url = url.replace('apis_extensions_v1beta1', 'api_v1')
+    # check if this is a subresource
+    subresource, resource_type, url = is_subresource(resource_type, url)
+    # get namespace name
+    name = url.split('api_v1_namespaces_').pop(1).split("_{}_".format(resource_type), 1).pop(0)
+    return 'api_v1_namespaces_' + name
+
+
+def is_subresource(resource_type, url):
+    # check if this is a subresource
+    subresource = resource_type
+    if url.endswith(resource_type):
+        # find the real resource type to do splitting below
+        url = url.replace('_' + resource_type, '')
+        match = list(set(resources) & set(url.split('_')))
+        match.remove('namespaces')  # not needed
+        # it is only a subresource when another resource can be found
+        if match:
+            resource_type = match.pop()
+
+    return subresource, resource_type, url
+
+
 @CacheLock()
 def process_hpa():
     """
@@ -570,7 +604,6 @@ def fetch_all(request, context):
     url = urlparse(request.url)
     filters = prepare_query_filters(url.query)
     cache_path = cache_key(request.path)
-
     data = filter_data(filters, cache_path)
     return {'items': data}
 
@@ -615,7 +648,9 @@ def get(request, context):
     """Process a GET request to the kubernetes API"""
     # Figure out if it is a GET operation for a single element or a list
     url = urlparse(request.url)
-    if get_type(url.path) in resources:
+    resource_type = get_type(url.path)
+    # is the last element a resource type or a name
+    if url.path.strip('/').split('/')[-1] == resource_type:
         return fetch_all(request, context)
 
     # fetch singular item
@@ -629,9 +664,7 @@ def post(request, context):
     resource_type = get_type(request.url)
     # check if the namespace being posted to exists
     if resource_type != 'namespaces':
-        namespace, _ = url.split('_{}_'.format(resource_type))
-        namespace = namespace.replace('apis_autoscaling_v1', 'api_v1')
-        namespace = namespace.replace('apis_extensions_v1beta1', 'api_v1')
+        namespace = get_namespace(url, resource_type)
         if cache.get(namespace) is None:
             context.status_code = 404
             context.reason = 'Not Found'
@@ -682,17 +715,21 @@ def put(request, context):
     """Process a PUT request to the kubernetes API"""
     url = cache_key(request.url)
     # type is the second last element
-    resource_type = get_type(request.url, -2)
+    resource_type = get_type(request.url)
     # check if the namespace being posted to exists
     if resource_type != 'namespaces':
-        namespace, _ = url.split('_{}_'.format(resource_type))
-        namespace = namespace.replace('apis_autoscaling_v1', 'api_v1')
-        namespace = namespace.replace('apis_extensions_v1beta1', 'api_v1')
+        namespace = get_namespace(url, resource_type)
         if cache.get(namespace) is None:
             context.status_code = 404
             context.reason = 'Not Found'
             return {}
 
+    # figure out main resource if in subresource
+    original_url = url
+    subresource, resource_type, url = is_subresource(resource_type, url)
+    if subresource != resource_type:
+        cache.set(original_url, request.json(), None)
+
     item = cache.get(url)
     if item is None:
         context.status_code = 404
@@ -702,14 +739,21 @@ def put(request, context):
     data = request.json()
 
     # merge new data into old but keep labels separate in case they changed
-    labels = data['metadata'].pop('labels')
-    item['metadata'].update(data['metadata'])
-    data['metadata'] = item['metadata']
-    # make sure only new labels are used
-    data['metadata']['labels'] = labels
+    if 'labels' in data['metadata']:
+        labels = data['metadata'].pop('labels')
+        item['metadata'].update(data['metadata'])
+        data['metadata'] = item['metadata']
+        # make sure only new labels are used
+        data['metadata']['labels'] = labels
 
     # split out deployments and replicasets? due to upsert_pods
     if resource_type in ['replicationcontrollers', 'replicasets', 'deployments']:
+        if subresource == 'scale':
+            # has minimal info so need to copy data
+            replicas = data['spec']['replicas']
+            data = copy.deepcopy(item)  # full copy
+            data['spec']['replicas'] = replicas
+
         if 'status' not in data:
             # just use what was set last time
             data['status'] = {'observedGeneration': item['status']['observedGeneration']}
@@ -744,7 +788,7 @@ def delete(request, context):
         context.reason = 'Not Found'
         return {}
 
-    resource_type = get_type(request.url, -2)
+    resource_type = get_type(request.url)
     # clean everything from a namespace
     if resource_type == 'namespaces':
         for resource in resources:
@@ -783,8 +827,8 @@ def add_cache_item(url, resource_type, data):
         cache.set(resource_type, items, None)
 
     # Keep track of what resources exist under other resources (mostly namespace)
-    namespace, item = url.split('_{}_'.format(resource_type))
-    cache_url = '{}_{}'.format(namespace, resource_type)
+    # sneaky way of getting data up to the resource type without too much magic
+    cache_url = ''.join(url.partition(resource_type)[0:2])
     items = cache.get(cache_url, [])
     if url not in items:
         items.append(url)
@@ -806,8 +850,8 @@ def remove_cache_item(url, resource_type):
         cache.set(resource_type, items, None)
 
     # remove from namespace specific scope
-    namespace, item = url.split('_{}_'.format(resource_type))
-    cache_url = '{}_{}'.format(namespace, resource_type)
+    # sneaky way of getting data up to the resource type without too much magic
+    cache_url = ''.join(url.partition(resource_type)[0:2])
     items = cache.get(cache_url, [])
     if url in items:
         items.remove(url)

rootfs/scheduler/resources/deployment.py

@@ -162,10 +162,14 @@ def scale(self, namespace, name, image, entrypoint, command, **kwargs):
             self.log(namespace, "Not scaling Deployment {} to {} replicas. Already at desired replicas".format(name, desired))  # noqa
             return
         elif desired != current:
+            self.log(namespace, "scaling Deployment {} from {} to {} replicas".format(name, current, desired))  # noqa
+            self.scales.update(namespace, name, desired, deployment)
+
+            # wait until scaling is done
+            self.wait_until_updated(namespace, name)
             # set the previous replicas count so the wait logic can deal with terminating pods
             kwargs['previous_replicas'] = current
-            self.log(namespace, "scaling Deployment {} from {} to {} replicas".format(name, current, desired))  # noqa
-            self.update(namespace, name, image, entrypoint, command, **kwargs)
+            self.wait_until_ready(namespace, name, **kwargs)
 
     def in_progress(self, namespace, name, deploy_timeout, batches, replicas, tags):
         """

rootfs/scheduler/resources/replicationcontroller.py

@@ -93,12 +93,8 @@ def scale(self, namespace, name, desired, timeout):
             self.log(namespace, "Not scaling RC {} to {} replicas. Already at desired replicas".format(name, desired))  # noqa
             return
         elif desired != rc['spec']['replicas']:  # RC needs new replica count
-            # Set the new desired replica count
-            rc['spec']['replicas'] = desired
-
             self.log(namespace, "scaling RC {} from {} to {} replicas".format(name, current, desired))  # noqa
-
-            self.update(namespace, name, rc)
+            self.scales.update(namespace, name, desired, rc)
             self.wait_until_updated(namespace, name)
 
         # Double check enough pods are in the required state to service the application

rootfs/scheduler/resources/scale.py

@@ -0,0 +1,37 @@
+from scheduler.resources import Resource
+from scheduler.exceptions import KubeHTTPException
+
+
+class Scale(Resource):
+    def manifest(self, namespace, name, replicas):
+        manifest = {
+            'kind': 'Scale',
+            'apiVersion': self.api_version,
+            'metadata': {
+                'namespace': namespace,
+                'name': name,
+            },
+            'spec': {
+                'replicas': replicas,
+            }
+        }
+
+        return manifest
+
+    def update(self, namespace, name, replicas, target):
+        # use API version and prefix from target use pick the right endpoint
+        resource_type = target['kind'].lower() + 's'  # make plural for url
+        self.api_version = getattr(self, resource_type).api_version
+        self.api_prefix = getattr(self, resource_type).api_prefix
+
+        manifest = self.manifest(namespace, name, replicas)
+        url = self.api("/namespaces/{}/{}/{}/scale", namespace, resource_type, name)
+        response = self.session.put(url, json=manifest)
+        if self.unhealthy(response.status_code):
+            raise KubeHTTPException(
+                response,
+                'scale {} "{}" in Namespace "{}"', target['kind'], name, namespace
+            )
+            self.log(namespace, 'template used: {}'.format(json.dumps(manifest, indent=4)), 'DEBUG')  # noqa
+
+        return response