feat(autoscale): add autoscaling support to application on per proc type basis (#1018)

helgi · web-flow · commit 5c83d80d3da7 · 2016-09-01T14:40:26.000-07:00
diff --git a/rootfs/api/migrations/0017_appsettings_autoscale.py b/rootfs/api/migrations/0017_appsettings_autoscale.py
@@ -0,0 +1,21 @@
+# -*- coding: utf-8 -*-
+# Generated by Django 1.10 on 2016-08-30 17:32
+from __future__ import unicode_literals
+
+from django.db import migrations
+import jsonfield.fields
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('api', '0016_auto_20160830_0104'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='appsettings',
+            name='autoscale',
+            field=jsonfield.fields.JSONField(blank=True, default={}),
+        ),
+    ]
diff --git a/rootfs/api/models/app.py b/rootfs/api/models/app.py
@@ -922,7 +922,7 @@ def _update_application_service(self, namespace, app_type, port, routable=False,
         except Exception as e:
             # Fix service to old port and app type
             self._scheduler.svc.update(namespace, namespace, data=old_service)
-            raise KubeException(str(e)) from e
+            raise ServiceUnavailable(str(e)) from e
 
     def whitelist(self, whitelist):
         """
@@ -936,3 +936,29 @@ def whitelist(self, whitelist):
             self._scheduler.svc.update(self.id, self.id, data=service)
         except KubeException as e:
             raise ServiceUnavailable(str(e)) from e
+
+    def autoscale(self, proc_type, autoscale):
+        """
+        Set autoscale rules for the application
+        """
+        name = '{}-{}'.format(self.id, proc_type)
+        # basically fake out a Deployment object (only thing we use) to assign to the HPA
+        target = {'kind': 'Deployment', 'metadata': {'name': name}}
+
+        try:
+            # get the target for autoscaler, in this case Deployment
+            self._scheduler.hpa.get(self.id, name)
+            if autoscale is None:
+                self._scheduler.hpa.delete(self.id, name)
+            else:
+                self._scheduler.hpa.update(
+                    self.id, name, proc_type, target, **autoscale
+                )
+        except KubeHTTPException as e:
+            if e.response.status_code == 404:
+                self._scheduler.hpa.create(
+                    self.id, name, proc_type, target, **autoscale
+                )
+            else:
+                # let the user know about any other errors
+                raise ServiceUnavailable(str(e)) from e
diff --git a/rootfs/api/models/appsettings.py b/rootfs/api/models/appsettings.py
@@ -2,9 +2,12 @@
 from django.conf import settings
 from django.db import models
 from django.contrib.postgres.fields import ArrayField
+from jsonfield import JSONField
+from rest_framework.exceptions import NotFound
 
+from api.utils import dict_diff
 from api.models import UuidAuditedModel
-from api.exceptions import DeisException, AlreadyExists
+from api.exceptions import DeisException, AlreadyExists, UnprocessableEntity
 
 
 class AppSettings(UuidAuditedModel):
@@ -17,6 +20,7 @@ class AppSettings(UuidAuditedModel):
     maintenance = models.NullBooleanField(default=None)
     routable = models.NullBooleanField(default=None)
     whitelist = ArrayField(models.CharField(max_length=50), default=[])
+    autoscale = JSONField(default={}, blank=True)
 
     class Meta:
         get_latest_by = 'created'
@@ -86,6 +90,46 @@ def update_whitelist(self, previous_settings):
                     self.summary += ' and '
                 self.summary += "{} {}".format(self.owner, changes)
 
+    def update_autoscale(self, previous_settings):
+        data = getattr(previous_settings, 'autoscale', {}).copy()
+        new = getattr(self, 'autoscale', {}).copy()
+        # If no previous settings then do nothing
+        if not previous_settings:
+            return
+
+        # if nothing changed copy the settings from previous
+        if not new and data:
+            setattr(self, 'autoscale', data)
+        elif data != new:
+            for proc, scale in new.items():
+                if scale is None:
+                    # error if unsetting non-existing key
+                    if proc not in data:
+                        raise UnprocessableEntity('{} does not exist under {}'.format(proc, 'autoscale'))  # noqa
+                    del data[proc]
+                else:
+                    data[proc] = scale
+            setattr(self, 'autoscale', data)
+
+            # only apply new items
+            for proc, scale in new.items():
+                self.app.autoscale(proc, scale)
+
+            # if the autoscale information changed, log the dict diff
+            changes = []
+            old_autoscale = getattr(previous_settings, 'autoscale', {})
+            diff = dict_diff(self.autoscale, old_autoscale)
+            # try to be as succinct as possible
+            added = ', '.join(list(map(lambda x: 'default' if x == '' else x, [k for k in diff.get('added', {})])))  # noqa
+            added = 'added autoscale for process type ' + added if added else ''
+            changed = ', '.join(list(map(lambda x: 'default' if x == '' else x, [k for k in diff.get('changed', {})])))  # noqa
+            changed = 'changed autoscale for process type ' + changed if changed else ''
+            deleted = ', '.join(list(map(lambda x: 'default' if x == '' else x, [k for k in diff.get('deleted', {})])))  # noqa
+            deleted = 'deleted autoscale for process type ' + deleted if deleted else ''
+            changes = ', '.join(i for i in (added, changed, deleted) if i)
+            if changes:
+                self.summary += ["{} {}".format(self.owner, changes)]
+
     def save(self, *args, **kwargs):
         self.summary = []
         previous_settings = None
@@ -98,6 +142,9 @@ def save(self, *args, **kwargs):
             self.update_maintenance(previous_settings)
             self.update_routable(previous_settings)
             self.update_whitelist(previous_settings)
+            self.update_autoscale(previous_settings)
+        except (UnprocessableEntity, NotFound):
+            raise
         except Exception as e:
             self.delete()
             raise DeisException(str(e)) from e
diff --git a/rootfs/api/serializers.py b/rootfs/api/serializers.py
@@ -470,6 +470,7 @@ class AppSettingsSerializer(serializers.ModelSerializer):
 
     app = serializers.SlugRelatedField(slug_field='id', queryset=models.App.objects.all())
     owner = serializers.ReadOnlyField(source='owner.username')
+    autoscale = JSONFieldSerializer(convert_to_str=False, required=False, binary=True)
 
     class Meta:
         """Metadata options for a :class:`AppSettingsSerializer`."""
@@ -489,6 +490,35 @@ def validate_whitelist(self, data):
                     except:
                         raise serializers.ValidationError(
                            "The address {} is not valid".format(address))
+
+        return data
+
+    def validate_autoscale(self, data):
+        schema = {
+            "$schema": "http://json-schema.org/schema#",
+            "type": "object",
+            "properties": {
+                # minimum replicas autoscale will keep resource at based on load
+                "min": {"type": "integer"},
+                # maximum replicas autoscale will keep resource at based on load
+                "max": {"type": "integer"},
+                # how much CPU load there is to trigger scaling rules
+                "cpu_percent": {"type": "integer"},
+            },
+            "required": ["min", "max", "cpu_percent"],
+        }
+
+        for proc, autoscale in data.items():
+            if autoscale is None:
+                continue
+
+            try:
+                jsonschema.validate(autoscale, schema)
+            except jsonschema.ValidationError as e:
+                raise serializers.ValidationError(
+                    "could not validate {}: {}".format(autoscale, e.message)
+                )
+
         return data
 
 
diff --git a/rootfs/api/tests/test_app_settings.py b/rootfs/api/tests/test_app_settings.py
@@ -7,10 +7,12 @@
 from api.models import App
 from unittest import mock
 from scheduler import KubeException
-from api.tests import adapter, DeisTransactionTestCase
+from api.tests import adapter, mock_port, DeisTransactionTestCase
 
 
 @requests_mock.Mocker(real_http=True, adapter=adapter)
+@mock.patch('api.models.release.publish_release', lambda *args: None)
+@mock.patch('api.models.release.docker_get_port', mock_port)
 class TestAppSettings(DeisTransactionTestCase):
     """Tests setting and updating config values"""
 
@@ -168,3 +170,83 @@ def test_kubernetes_service_failure(self, mock_requests):
             url = '/v2/apps/{}/whitelist'.format(app_id)
             response = self.client.post(url, {'addresses': addresses})
             self.assertEqual(response.status_code, 400, response.data)
+
+    def test_autoscale(self, mock_requests):
+        """
+        Test that autoscale can be applied
+        """
+        app_id = self.create_app()
+
+        # create an autoscaling rule
+        scale = {'autoscale': {'cmd': {'min': 2, 'max': 5, 'cpu_percent': 45}}}
+        response = self.client.post(
+            '/v2/apps/{app_id}/settings'.format(**locals()),
+            scale
+        )
+        self.assertEqual(response.status_code, 201, response.data)
+        self.assertIn('cmd', response.data['autoscale'])
+        self.assertEqual(response.data['autoscale'], scale['autoscale'])
+
+        # update
+        scale = {'autoscale': {'cmd': {'min': 2, 'max': 8, 'cpu_percent': 45}}}
+        response = self.client.post(
+            '/v2/apps/{app_id}/settings'.format(**locals()),
+            scale
+        )
+        self.assertEqual(response.status_code, 201, response.data)
+        self.assertIn('cmd', response.data['autoscale'])
+        self.assertEqual(response.data['autoscale'], scale['autoscale'])
+
+        # create
+        scale = {'autoscale': {'worker': {'min': 2, 'max': 5, 'cpu_percent': 45}}}
+        response = self.client.post(
+            '/v2/apps/{app_id}/settings'.format(**locals()),
+            scale
+        )
+        self.assertEqual(response.status_code, 201, response.data)
+        self.assertIn('worker', response.data['autoscale'])
+        self.assertEqual(response.data['autoscale']['worker'], scale['autoscale']['worker'])
+
+        # check that the cmd proc type is still there
+        self.assertIn('cmd', response.data['autoscale'])
+
+        # check that config fails if trying to unset non-existing proc type
+        response = self.client.post(
+            '/v2/apps/{app_id}/settings'.format(**locals()),
+            {'autoscale': {'invalid_proctype': None}})
+        self.assertEqual(response.status_code, 422, response.data)
+
+        # remove a proc type
+        response = self.client.post(
+            '/v2/apps/{app_id}/settings'.format(**locals()),
+            {'autoscale': {'worker': None}})
+        self.assertEqual(response.status_code, 201, response.data)
+        self.assertNotIn('worker', response.data['autoscale'])
+        self.assertIn('cmd', response.data['autoscale'])
+
+        # remove another proc type
+        response = self.client.post(
+            '/v2/apps/{app_id}/settings'.format(**locals()),
+            {'autoscale': {'cmd': None}})
+        self.assertEqual(response.status_code, 201, response.data)
+        self.assertNotIn('cmd', response.data['autoscale'])
+
+    def test_autoscale_validations(self, mock_requests):
+        """
+        Test that autoscale validations work
+        """
+        app_id = self.create_app()
+
+        # Set one of the values that require a numeric value to a string
+        response = self.client.post(
+            '/v2/apps/{app_id}/settings'.format(**locals()),
+            {'autoscale': {'cmd': {'min': 4, 'max': 5, 'cpu_percent': "t"}}}
+        )
+        self.assertEqual(response.status_code, 400, response.data)
+
+        # Don't set one of the mandatory value
+        response = self.client.post(
+            '/v2/apps/{app_id}/settings'.format(**locals()),
+            {'autoscale': {'cmd': {'min': 4, 'cpu_percent': 45}}}
+        )
+        self.assertEqual(response.status_code, 400, response.data)
diff --git a/rootfs/scheduler/mock.py b/rootfs/scheduler/mock.py
@@ -124,8 +124,12 @@ def process_hpa():
         hpa = cache.get(row)
 
         # check if the resource referenced actually exists
-        kind = hpa['spec']['scaleRef']['kind'].lower() + 's'  # make plural
-        name = hpa['spec']['scaleRef']['name'].lower()
+        if 'scaleRef' in hpa['spec']:
+            kind = hpa['spec']['scaleRef']['kind'].lower() + 's'  # make plural
+            name = hpa['spec']['scaleRef']['name'].lower()
+        else:
+            kind = hpa['spec']['scaleTargetRef']['kind'].lower() + 's'  # make plural
+            name = hpa['spec']['scaleTargetRef']['name'].lower()
         deployment = None
         for deploy in cache.get(kind, []):
             item = cache.get(deploy)
diff --git a/rootfs/scheduler/resources/horizontalpodautoscaler.py b/rootfs/scheduler/resources/horizontalpodautoscaler.py
@@ -38,7 +38,7 @@ def get(self, namespace, name=None, **kwargs):
 
         return response
 
-    def manifest(self, namespace, name, target, **kwargs):
+    def manifest(self, namespace, name, app_type, target, **kwargs):
         min_replicas = kwargs.get('min')
         max_replicas = kwargs.get('max')
         cpu_percent = kwargs.get('cpu_percent')
@@ -51,7 +51,7 @@ def manifest(self, namespace, name, target, **kwargs):
 
         labels = {
             'app': namespace,
-            'type': kwargs.get('app_type'),
+            'type': app_type,
             'heritage': 'deis',
         }
 
@@ -66,41 +66,53 @@ def manifest(self, namespace, name, target, **kwargs):
             'spec': {
                 'minReplicas': min_replicas,
                 'maxReplicas': max_replicas,
-                'scaleRef': {
-                    # only works with Deployments, RS and RC
-                    'kind': target['kind'],
-                    'name': target['metadata']['name'],
-                    # the resource of the above which does the scale action
-                    'subresource': 'scale',
-                },
-                'cpuUtilization': {
-                    'targetPercentage': cpu_percent
-                }
             }
         }
 
+        if self.version() >= 1.3:
+            manifest['spec']['targetCPUUtilizationPercentage'] = cpu_percent
+
+            manifest['spec']['scaleTargetRef'] = {
+                # only works with Deployments, RS and RC
+                'kind': target['kind'],
+                'name': target['metadata']['name'],
+            }
+        elif self.version() <= 1.2:
+            # api changed between version
+            manifest['spec']['cpuUtilization'] = {
+                'targetPercentage': cpu_percent
+            }
+
+            manifest['spec']['scaleRef'] = {
+                # only works with Deployments, RS and RC
+                'kind': target['kind'],
+                'name': target['metadata']['name'],
+                # the resource of the above which does the scale action
+                'subresource': 'scale',
+            }
+
         return manifest
 
-    def create(self, namespace, name, target, **kwargs):
-        manifest = self.manifest(namespace, name, target, **kwargs)
+    def create(self, namespace, name, app_type, target, **kwargs):
+        manifest = self.manifest(namespace, name, app_type, target, **kwargs)
 
         url = self.api("/namespaces/{}/horizontalpodautoscalers", namespace)
         response = self.session.post(url, json=manifest)
         if self.unhealthy(response.status_code):
+            self.log(namespace, 'template used: {}'.format(json.dumps(manifest, indent=4)), 'DEBUG')  # noqa
             raise KubeHTTPException(
                 response,
                 'create HorizontalPodAutoscaler "{}" in Namespace "{}"', name, namespace
             )
-            self.log(namespace, 'template used: {}'.format(json.dumps(manifest, indent=4)), 'DEBUG')  # noqa
 
         # optionally wait for HPA if requested
         if kwargs.get('wait', False):
             self.wait(namespace, name)
 
         return response
 
-    def update(self, namespace, name, target, **kwargs):
-        manifest = self.manifest(namespace, name, target, **kwargs)
+    def update(self, namespace, name, app_type, target, **kwargs):
+        manifest = self.manifest(namespace, name, app_type, target, **kwargs)
 
         url = self.api("/namespaces/{}/horizontalpodautoscalers/{}", namespace, name)
         response = self.session.put(url, json=manifest)
@@ -137,8 +149,12 @@ def wait(self, namespace, name):
         # ideally it would use the resources wait commands but they vary
         for _ in range(30):
             # fetch resource attached to it
-            resource_kind = hpa['spec']['scaleRef']['kind'].lower()
-            resource_name = hpa['spec']['scaleRef']['name']
+            if self.version() >= 1.3:
+                resource_kind = hpa['spec']['scaleTargetRef']['kind'].lower()
+                resource_name = hpa['spec']['scaleTargetRef']['name']
+            elif self.version() <= 1.2:
+                resource_kind = hpa['spec']['scaleRef']['kind'].lower()
+                resource_name = hpa['spec']['scaleRef']['name']
 
             resource = getattr(self, resource_kind)
             resource = getattr(resource, 'get')(namespace, resource_name).json()
diff --git a/rootfs/scheduler/tests/test_horizontalpodautoscaler.py b/rootfs/scheduler/tests/test_horizontalpodautoscaler.py
