feat(k8s): integrate kubernetes scheduler with deis.

Author: Keerthan Mala

contrib/coreos/user-data.example

@@ -159,6 +159,9 @@ write_files:
   - path: /etc/deis-release
     content: |
       DEIS_RELEASE=v1.8.0
+  - path: /etc/k8environment
+    content: |
+      K8S_VERSION=v0.21.0
   - path: /etc/motd
     content: " \e[31m* *    \e[34m*   \e[32m*****    \e[39mddddd   eeeeeee iiiiiii   ssss\n\e[31m*   *  \e[34m* *  \e[32m*   *     \e[39md   d   e    e    i     s    s\n \e[31m* *  \e[34m***** \e[32m*****     \e[39md    d  e         i    s\n\e[32m*****  \e[31m* *    \e[34m*       \e[39md     d e         i     s\n\e[32m*   * \e[31m*   *  \e[34m* *      \e[39md     d eee       i      sss\n\e[32m*****  \e[31m* *  \e[34m*****     \e[39md     d e         i         s\n  \e[34m*   \e[32m*****  \e[31m* *      \e[39md    d  e         i          s\n \e[34m* *  \e[32m*   * \e[31m*   *     \e[39md   d   e    e    i    s    s\n\e[34m***** \e[32m*****  \e[31m* *     \e[39mddddd   eeeeeee iiiiiii  ssss\n\n\e[39mWelcome to Deis\t\t\tPowered by Core\e[38;5;45mO\e[38;5;206mS\e[39m\n"
   - path: /etc/profile.d/nse-function.sh
@@ -316,4 +319,38 @@ write_files:
       # removing the node from etcd
       NODE=$($ETCDCTL member list | grep `cat /etc/machine-id` | cut -d ':' -f 1)
       $ETCDCTL member remove $NODE
+  - path: /opt/bin/download-k8s-binary
+    permissions: '0755'
+    content: |
+      #!/bin/bash
+
+      . /etc/k8environment
+
+      mkdir -p /opt/bin
+
+      FILE=$1
+      if [ ! -f /opt/bin/$FILE ]; then
+        curl -sSL -o /opt/bin/$FILE https://storage.googleapis.com/kubernetes-release/release/${K8S_VERSION}/bin/linux/amd64/$FILE
+        chmod +x /opt/bin/$FILE
+      else
+        # we check the version of the binary
+        INSTALLED_VERSION=$(/opt/bin/$FILE --version)
+        MATCH=$(echo "${INSTALLED_VERSION}" | grep -c "${K8S_VERSION}")
+        if [ $MATCH -eq 0 ]; then
+          # the version is different
+          curl -sSL -o /opt/bin/$FILE https://storage.googleapis.com/kubernetes-release/release/${K8S_VERSION}/bin/linux/amd64/$FILE
+          chmod +x /opt/bin/$FILE
+        fi
+      fi
+  - path: /opt/bin/kube-serviceaccount
+    permissions: '0755'
+    content: |
+      #!/bin/bash
+
+      if etcdctl get /kube-serviceaccount >/dev/null 2>&1; then
+        etcdctl get /kube-serviceaccount > /opt/bin/kube-serviceaccount.key
+      else
+        /bin/openssl genrsa -out /opt/bin/kube-serviceaccount.key 2048 2>/dev/null
+        etcdctl set /kube-serviceaccount < /opt/bin/kube-serviceaccount.key
+      fi
 manage_etc_hosts: localhost

controller/api/models.py

@@ -179,6 +179,28 @@ def __str__(self):
     def url(self):
         return self.id + '.' + settings.DEIS_DOMAIN
 
+    def _get_job_id(self,container_type):
+        app = self.id
+        release = self.release_set.latest()
+        version = "v{}".format(release.version)
+        job_id = "{app}_{version}.{container_type}".format(**locals())
+        return job_id
+
+    def _get_command(self,container_type):
+        try:
+            # if this is not procfile-based app, ensure they cannot break out
+            # and run arbitrary commands on the host
+            # FIXME: remove slugrunner's hardcoded entrypoint
+            release = self.release_set.latest()
+            if release.build.dockerfile or not release.build.sha:
+                return "bash -c '{}'".format(release.build.procfile[container_type])
+            else:
+                return 'start {}'.format(container_type)
+        # if the key is not present or if a parent attribute is None
+        except (KeyError, TypeError, AttributeError):
+            # handle special case for Dockerfile deployments
+            return '' if container_type == 'cmd' else 'start {}'.format(container_type)
+
     def log(self, message):
         """Logs a message to the application's log file.
 
@@ -242,6 +264,8 @@ def scale(self, user, structure):  # noqa
         # iterate and scale by container type (web, worker, etc)
         changed = False
         to_add, to_remove = [], []
+        scale_types = {}
+
         # iterate on a copy of the container_type keys
         for container_type in requested_structure.keys():
             containers = list(self.container_set.filter(type=container_type).order_by('created'))
@@ -253,20 +277,11 @@ def scale(self, user, structure):  # noqa
             if diff == 0:
                 continue
             changed = True
+            scale_types[container_type] = requested
             while diff < 0:
                 c = containers.pop()
-                if settings.SCHEDULER_MODULE == "scheduler.k8s" :
-                    if len(containers) == 0:
-                        t = Thread(target=c.scale)
-                        t.start()
-                        t.join()
-                    c.delete()
-                else:
-                    to_remove.append(c)
+                to_remove.append(c)
                 diff += 1
-            if diff == 0 and len(containers) != 0:
-                c = containers.pop()
-                self._start_containers([c])
             while diff > 0:
                 # create a database record
                 c = Container.objects.create(owner=self.owner,
@@ -277,14 +292,15 @@ def scale(self, user, structure):  # noqa
                 to_add.append(c)
                 container_num += 1
                 diff -= 1
+
         if changed:
-            if to_add:
-                if settings.SCHEDULER_MODULE == "scheduler.k8s" :
-                    self._start_containers([to_add.pop()])
-                else:
+            if "scale" in dir(self._scheduler) :
+                self._scale_containers(scale_types,to_remove)
+            else :
+                if to_add:
                     self._start_containers(to_add)
-            if to_remove:
-                self._destroy_containers(to_remove)
+                if to_remove:
+                    self._destroy_containers(to_remove)
         # save new structure to the database
         vals = self.container_set.exclude(type='run').values(
             'type').annotate(Count('pk')).order_by()
@@ -294,6 +310,31 @@ def scale(self, user, structure):  # noqa
         self.save()
         return changed
 
+    def _scale_containers(self,scale_types,to_remove) :
+        release = self.release_set.latest()
+        for scale_type in scale_types :
+            image = release.image
+            version = "v{}".format(release.version)
+            kwargs = {'memory': release.config.memory,
+                      'cpu': release.config.cpu,
+                      'tags': release.config.tags,
+                      'version': version,
+                      'aname': self.id,
+                      'num': scale_types[scale_type]}
+            job_id = self._get_job_id(scale_type)
+            command = self._get_command(scale_type)
+            try:
+                self._scheduler.scale(
+                    name=job_id,
+                    image=image,
+                    command=command,
+                    **kwargs)
+            except Exception as e:
+                err = '{} (scale): {}'.format(job_id, e)
+                log_event(self, err, logging.ERROR)
+                raise
+        [c.delete() for c in to_remove]
+
     def _start_containers(self, to_add):
         """Creates and starts containers via the scheduler"""
         if not to_add:
@@ -302,15 +343,15 @@ def _start_containers(self, to_add):
         start_threads = [Thread(target=c.start) for c in to_add]
         [t.start() for t in create_threads]
         [t.join() for t in create_threads]
-        if settings.SCHEDULER_MODULE != "scheduler.k8s" and any(c.state != 'created' for c in to_add):
-            err = 'aborting, failed to create some containers '+c.state
+        if any(c.state != 'created' for c in to_add):
+            err = 'aborting, failed to create some containers'
             log_event(self, err, logging.ERROR)
             self._destroy_containers(to_add)
             raise RuntimeError(err)
         [t.start() for t in start_threads]
         [t.join() for t in start_threads]
         if set([c.state for c in to_add]) != set(['up']):
-            err = 'warning, some containers failed to start'+c.state
+            err = 'warning, some containers failed to start'
             log_event(self, err, logging.WARNING)
 
     def _restart_containers(self, to_restart):
@@ -334,43 +375,64 @@ def _destroy_containers(self, to_destroy):
         """Destroys containers via the scheduler"""
         if not to_destroy:
             return
-        if settings.SCHEDULER_MODULE == "scheduler.k8s" :
-            destroy_threads = [Thread(target=to_destroy[0].destroy)]
-        else:
-            destroy_threads = [Thread(target=c.destroy) for c in to_destroy]
+        destroy_threads = [Thread(target=c.destroy) for c in to_destroy]
         [t.start() for t in destroy_threads]
         [t.join() for t in destroy_threads]
-        if settings.SCHEDULER_MODULE == "scheduler.k8s" :
-            [c.delete() for c in to_destroy]
-        else :
-            [c.delete() for c in to_destroy if c.state == 'destroyed']
+        [c.delete() for c in to_destroy if c.state == 'destroyed']
         if any(c.state != 'destroyed' for c in to_destroy):
-            err = 'aborting, failed to destroy some containers'+c.state
+            err = 'aborting, failed to destroy some containers'
             log_event(self, err, logging.ERROR)
             raise RuntimeError(err)
 
     def deploy(self, user, release):
         """Deploy a new release to this application"""
         existing = self.container_set.exclude(type='run')
         new = []
+        scale_types = set()
+        old_name = ''
         for e in existing:
             n = e.clone(release)
             n.save()
             new.append(n)
+            scale_types.add(e.type)
 
-        if new and settings.SCHEDULER_MODULE == "scheduler.k8s" :
-            self._start_containers([new.pop()])
+        if new and "deploy" in dir(self._scheduler) :
+            self._deploy_app(scale_types,release,existing)
         else :
             self._start_containers(new)
 
-        # destroy old containers
-        if existing:
-            self._destroy_containers(existing)
+            # destroy old containers
+            if existing:
+                self._destroy_containers(existing)
 
         # perform default scaling if necessary
         if self.structure == {} and release.build is not None:
             self._default_scale(user, release)
 
+    def _deploy_app(self,scale_types,release,existing) :
+        for scale_type in scale_types :
+            image = release.image
+            version = "v{}".format(release.version)
+            kwargs = {'memory': release.config.memory,
+                      'cpu': release.config.cpu,
+                      'tags': release.config.tags,
+                      'aname': self.id,
+                      'num': 0,
+                      'version':version}
+            job_id = self._get_job_id(scale_type)
+            command = self._get_command(scale_type)
+            try:
+                self._scheduler.deploy(
+                    name=job_id,
+                    image=image,
+                    command=command,
+                    **kwargs)
+            except Exception as e:
+                err = '{} (deploy): {}'.format(job_id, e)
+                log_event(self, err, logging.ERROR)
+                raise
+        [c.delete() for c in existing]
+
     def _default_scale(self, user, release):
         """Scale to default structure based on release type"""
         # if there is no SHA, assume a docker image is being promoted
@@ -504,9 +566,7 @@ def create(self):
         image = self.release.image
         kwargs = {'memory': self.release.config.memory,
                   'cpu': self.release.config.cpu,
-                  'tags': self.release.config.tags,
-                  'aname': self.app.id,
-                  'num': self.num}
+                  'tags': self.release.config.tags}
         job_id = self._job_id
         try:
             self._scheduler.create(
@@ -519,20 +579,6 @@ def create(self):
             log_event(self.app, err, logging.ERROR)
             raise
 
-    @close_db_connections
-    def scale(self):
-        image = self.release.image
-        job_id = self._job_id
-        try:
-            self._scheduler.scale(
-                name=job_id,
-                image=image,
-                num=0)
-        except Exception as e:
-            err = '{} (scale): {}'.format(job_id, e)
-            log_event(self.app, err, logging.ERROR)
-            raise
-
     @close_db_connections
     def start(self):
         job_id = self._job_id