Merge pull request #89 from helgi/domains_k8s

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

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