ref(models): add domains to k8s data for router to use

Domain information is now saved to routerConfig annotations in addition to still being in etcd. This solves part of #50
Also adds new flatten and dict_merge to both scheduler and models

The code has a little oddity in it where subdomains for app (on the platform domain) are added during scale event from 0 to 1. That will be address in #88
It is possible to remove the apps subdomain but during any scaling event it gets added back to the Domain model. This allows the router to treat all domains the same way

Author: helgi

rootfs/api/models.py

@@ -12,6 +12,7 @@
 import logging
 import re
 import time
+import json
 from threading import Thread
 
 from django.conf import settings
@@ -30,7 +31,7 @@
 
 from api import fields, utils, exceptions
 from registry import publish_release
-from utils import dict_diff, fingerprint
+from utils import dict_diff, dict_merge, fingerprint
 
 
 logger = logging.getLogger(__name__)
@@ -325,7 +326,13 @@ def _scale_containers(self, scale_types, to_remove):
                     name=job_id,
                     image=image,
                     command=command,
-                    **kwargs)
+                    **kwargs
+                )
+
+                # Attach the platform specific application sub domain
+                # scheduler.scale creates the required service on apps:create
+                if not Domain.objects.filter(owner=self.owner, app=self, domain=self).exists():
+                    Domain(owner=self.owner, app=self, domain=str(self)).save()
             except Exception as e:
                 err = '{} (scale): {}'.format(job_id, e)
                 log_event(self, err, logging.ERROR)
@@ -614,7 +621,8 @@ def create(self):
                 name=self.job_id,
                 image=image,
                 command=self._command,
-                **kwargs)
+                **kwargs
+            )
         except Exception as e:
             err = '{} (create): {}'.format(self.job_id, e)
             log_event(self.app, err, logging.ERROR)
@@ -965,6 +973,69 @@ class Domain(AuditedModel):
     app = models.ForeignKey('App')
     domain = models.TextField(blank=False, null=False, unique=True)
 
+    @property
+    def _scheduler(self):
+        mod = importlib.import_module(settings.SCHEDULER_MODULE)
+        return mod.SchedulerClient(settings.SCHEDULER_URL,
+                                   settings.SCHEDULER_AUTH,
+                                   settings.SCHEDULER_OPTIONS)
+
+    def _load_service_config(self, app):
+        # Get the service from k8s to attach the domain correctly
+        svc = self._scheduler._get_service(app, app).json()
+        # Get minimum structure going if it is missing on the service
+        if 'metadata' not in svc or 'annotations' not in svc['metadata']:
+            default = {'metadata': {'annotations': {}}}
+            svc = dict_merge(svc, default)
+
+        # Check if any config has been set
+        if 'deis.io/routerConfig' not in svc['metadata']['annotations']:
+            config = {}
+        else:
+            config = json.loads(svc['metadata']['annotations']['deis.io/routerConfig'])
+
+        # See if domains are available
+        if 'domains' not in config:
+            config['domains'] = []
+
+        return svc, config
+
+    def save(self, *args, **kwargs):
+        app = str(self.app)
+        domain = str(self.domain)
+
+        # setup the service and config dict
+        svc, config = self._load_service_config(app)
+        if domain not in config['domains']:
+            config['domains'].append(domain)
+
+        # save as a JSON string since annotations don't take a structure on its keys
+        svc['metadata']['annotations']['deis.io/routerConfig'] = json.dumps(config)
+
+        # Update the k8s service for the application with new domain information
+        self._scheduler._update_service(app, app, svc)
+
+        # Save to DB
+        return super(Domain, self).save(*args, **kwargs)
+
+    def delete(self, *args, **kwargs):
+        app = str(self.app)
+        domain = str(self.domain)
+
+        # setup the service and config dict
+        svc, config = self._load_service_config(app)
+        if domain in config['domains']:
+            config['domains'].remove(domain)
+
+        # save as a JSON string since annotations don't take a structure on its keys
+        svc['metadata']['annotations']['deis.io/routerConfig'] = json.dumps(config)
+
+        # Update the k8s service for the application with new domain information
+        self._scheduler._update_service(app, app, svc)
+
+        # Delete from DB
+        return super(Domain, self).delete(*args, **kwargs)
+
     def __str__(self):
         return self.domain
 
@@ -1154,21 +1225,6 @@ def _etcd_purge_config(**kwargs):
     except KeyError:
         pass
 
-
-def _etcd_publish_domains(**kwargs):
-    domain = kwargs['instance']
-    _etcd_client.write('/deis/domains/{}'.format(domain), domain.app)
-
-
-def _etcd_purge_domains(**kwargs):
-    domain = kwargs['instance']
-    try:
-        _etcd_client.delete('/deis/domains/{}'.format(domain),
-                            prevExist=True, dir=True, recursive=True)
-    except KeyError:
-        pass
-
-
 # Log significant app-related events
 post_save.connect(_log_build_created, sender=Build, dispatch_uid='api.models.log')
 post_save.connect(_log_release_created, sender=Release, dispatch_uid='api.models.log')
@@ -1193,8 +1249,6 @@ def create_auth_token(sender, instance=None, created=False, **kwargs):
     post_save.connect(_etcd_publish_key, sender=Key, dispatch_uid='api.models')
     post_delete.connect(_etcd_purge_key, sender=Key, dispatch_uid='api.models')
     post_delete.connect(_etcd_purge_user, sender=get_user_model(), dispatch_uid='api.models')
-    post_save.connect(_etcd_publish_domains, sender=Domain, dispatch_uid='api.models')
-    post_delete.connect(_etcd_purge_domains, sender=Domain, dispatch_uid='api.models')
     post_save.connect(_etcd_publish_app, sender=App, dispatch_uid='api.models')
     post_delete.connect(_etcd_purge_app, sender=App, dispatch_uid='api.models')
     post_save.connect(_etcd_publish_cert, sender=Certificate, dispatch_uid='api.models')

rootfs/api/tests/test_utils.py

@@ -0,0 +1,78 @@
+import unittest
+from api import utils
+
+
+class TestUtils(unittest.TestCase):
+    """Test utils functions"""
+
+    def test_dict_merge_simple(self):
+        a = {'key': 'value'}
+        b = {'key': 'value'}
+
+        c = utils.dict_merge(a, b)
+        assert c == {'key': 'value'}
+
+        a = {'key': 'value'}
+        b = {'key2': 'value'}
+
+        c = utils.dict_merge(a, b)
+        assert c == {'key': 'value', 'key2': 'value'}
+
+    def test_dict_merge_deeper(self):
+        a = {'key': 'value', 'here': {'without': 'you'}}
+        b = {'this': 'that', 'here': {'with': 'me'}, 'other': {'magic', 'unicorn'}}
+
+        c = utils.dict_merge(a, b)
+        assert c == {
+            'key': 'value',
+            'this': 'that',
+            'here': {
+                'with': 'me',
+                'without': 'you'
+            },
+            'other': {'magic', 'unicorn'}
+        }
+
+    def test_dict_merge_even_deeper(self):
+        a = {
+            'key': 'value',
+            'here': {'without': 'you'},
+            'other': {'scrubs': {'char3': 'Cox'}}
+
+        }
+
+        b = {
+            'this': 'that',
+            'here': {'with': 'me'},
+            'other': {'magic': 'unicorn', 'scrubs': {'char1': 'JD', 'char2': 'Turk'}}
+        }
+
+        c = utils.dict_merge(a, b)
+        assert c == {
+            'key': 'value',
+            'this': 'that',
+            'here': {'with': 'me', 'without': 'you'},
+            'other': {
+                'magic': 'unicorn',
+                'scrubs': {
+                    'char1': 'JD',
+                    'char2': 'Turk',
+                    'char3': 'Cox'
+                }
+            }
+        }
+
+    def test_dict_merge_with_list(self):
+        a = {'key': 'value', 'names': ['bob', 'kyle', 'kenny', 'jimbo']}
+        b = {'key': 'value', 'names': ['kenny', 'cartman', 'stan']}
+
+        c = utils.dict_merge(a, b)
+        assert c == {'key': 'value', 'names': ['bob', 'kyle', 'kenny', 'jimbo', 'cartman', 'stan']}
+
+    def test_dict_merge_bad_merge(self):
+        """Returns b because it isn't a dict"""
+        a = {'key': 'value'}
+        b = 'duh'
+
+        c = utils.dict_merge(a, b)
+        assert c == b

rootfs/api/utils.py

@@ -1,9 +1,11 @@
 """
 Helper functions used by the Deis server.
 """
+from __future__ import unicode_literals
 import base64
 import hashlib
 import random
+from copy import deepcopy
 
 
 def generate_app_name():
@@ -116,6 +118,57 @@ def encode(obj):
         return obj
 
 
+def dict_merge(origin, merge):
+    """
+    Recursively merges dict's. not just simple a["key"] = b["key"], if
+    both a and b have a key who's value is a dict then dict_merge is called
+    on both values and the result stored in the returned dictionary.
+    Also handles merging lists if they occur within the dict
+    """
+    if not isinstance(merge, dict):
+        return merge
+
+    result = deepcopy(origin)
+    for key, value in merge.iteritems():
+        if key in result and isinstance(result[key], dict):
+            result[key] = dict_merge(result[key], value)
+        else:
+            if isinstance(value, list):
+                if key not in result:
+                    result[key] = value
+                else:
+                    # merge lists without leaving potential duplicates
+                    # result[key] = list(set(result[key] + value))  # changes the order as well
+                    for item in value:
+                        if item in result[key]:
+                            continue
+
+                        result[key].append(item)
+            else:
+                result[key] = deepcopy(value)
+    return result
+
+
+def flatten(collection):
+    """
+    Flatten an arbitrarily deep structure of dicts, lists and tuples and
+    extract the strings at the leaves of the structures.
+    """
+    vals = []
+    if isinstance(collection, basestring):
+        vals.append(collection)
+    elif isinstance(collection, (list, tuple)):
+        for item in collection:
+            vals.extend(flatten(item))
+    elif isinstance(collection, dict):
+        vals.extend(flatten(collection.values()))
+    else:
+        raise Exception(
+            "Can't extract values from a %s" % collection.__class__
+        )
+    return vals
+
+
 if __name__ == "__main__":
     import doctest
     doctest.testmod()

rootfs/scheduler/k8s.py

@@ -1,3 +1,4 @@
+from __future__ import unicode_literals
 import json
 import logging
 import re
@@ -10,6 +11,8 @@
 from .states import JobState
 from . import AbstractSchedulerClient
 import requests
+from .utils import dict_merge
+
 
 logger = logging.getLogger(__name__)
 
@@ -82,7 +85,8 @@
     "name": "$name",
     "labels": {
       "app": "$name"
-    }
+    },
+    "annotations": {}
   },
   "spec": {
     "ports": [
@@ -392,25 +396,32 @@ def create(self, name, image, command, **kwargs):
         name = name.replace('.', '-').replace('_', '-')
         app_name = kwargs.get('aname', {})
         try:
-            self._create_service(name, app_name, app_type)
+            # Make sure the router knows what to do with this
+            data = {}
+            # TODO this should potentially be higher up in the flow
+            # see http://docs.deis.io/en/latest/using_deis/process-types/#web-vs-cmd-process-types
+            if app_type in ['web', 'cmd']:
+                data = {'metadata': {'labels': {'routable': 'true'}}}
+
+            self._create_service(name, app_name, app_type, data)
         except:
             self._scale_app(name, 0, app_name)
             self._delete_rc(name, app_name)
             raise
 
     def _get_service(self, name, namespace):
         url = self._api("/namespaces/{}/services/{}", namespace, name)
-        resp = self.session.get(url)
-        if unhealthy(resp.status_code):
-            error(resp, 'get Service "{}" in Namespace "{}"', name, namespace)
+        response = self.session.get(url)
+        if unhealthy(response.status_code):
+            error(response, 'get Service "{}" in Namespace "{}"', name, namespace)
 
-        return resp.status_code, resp.text, resp.reason
+        return response
 
-    def _create_service(self, name, app_name, app_type):
+    def _create_service(self, name, app_name, app_type, data={}):
         actual_pod = {}
         for _ in xrange(300):
-            status, data, reason = self._get_pods(app_name)
-            parsed_json = json.loads(data)
+            status, json_data, reason = self._get_pods(app_name)
+            parsed_json = json.loads(json_data)
             for pod in parsed_json['items']:
                 if('generateName' in pod['metadata'] and
                    pod['metadata']['generateName'] == name + '-'):
@@ -441,24 +452,29 @@ def _create_service(self, name, app_name, app_type):
             "name": app_name,
         }
 
-        template = string.Template(SERVICE_TEMPLATE).substitute(l)
+        # Merge external data on to the prefined template
+        template = json.loads(string.Template(SERVICE_TEMPLATE).substitute(l))
+        data = dict_merge(template, data)
+
         url = self._api("/namespaces/{}/services", app_name)
-        resp = self.session.post(url, json=json.loads(template))
+        resp = self.session.post(url, json=data)
         if resp.status_code == 409:
-            status, data, reason = self._get_service(app_name, app_name)
-            srv = json.loads(data)
+            srv = self._get_service(app_name, app_name).json()
             if srv['spec']['selector']['type'] == 'web':
                 return
 
             srv['spec']['selector']['type'] = app_type
             srv['spec']['ports'][0]['targetPort'] = port
-            url = self._api("/namespaces/{}/services/{}", app_name, app_name)
-            resp2 = self.session.put(url, json=srv)
+            resp2 = self._update_service(app_name, app_name, srv)
             if unhealthy(resp2.status_code):
                 error(resp, 'update Service "{}" in Namespace "{}"', app_name, app_name)
         elif unhealthy(resp.status_code):
             error(resp, 'create Service "{}" in Namespace "{}"', app_name, app_name)
 
+    def _update_service(self, namespace, app, data):
+        url = self._api("/namespaces/{}/services/{}", namespace, app)
+        return self.session.put(url, json=data)
+
     def start(self, name):
         """Start a container."""
         pass