Merge pull request #3975 from kmala/k8s

feat(k8s) : add k8s scheduler support to deis

Author: mboersma

contrib/coreos/user-data.example

@@ -316,4 +316,42 @@ 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/wupiao
+    permissions: '0755'
+    content: |
+      #!/bin/bash
+      # [w]ait [u]ntil [p]ort [i]s [a]ctually [o]pen
+      [ -n "$1" ] && \
+        until curl -o /dev/null -sIf http://${1}; do \
+          sleep 1 && echo .;
+        done;
+      exit $?
+  - path: /opt/bin/download-k8s-binary
+    permissions: '0755'
+    content: |
+      #!/bin/bash
+      export K8S_VERSION="v1.0.1"
+      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/scheduler-policy.json
+    content: |
+      {
+          "kind": "Policy",
+          "apiVersion": "v1",
+          "predicates": [{"name": "PodFitsPorts"},{"name": "PodFitsResources"},{"name": "NoDiskConflict"},{"name": "MatchNodeSelector"},{"name": "HostName"}],
+          "priorities": [{"name": "LeastRequestedPriority","weight": 1},{"name": "BalancedResourceAllocation","weight": 1},{"name": "ServiceSpreadingPriority","weight": 2},{"name": "EqualPriority","weight": 1}]
+      }
 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,6 +277,7 @@ def scale(self, user, structure):  # noqa
             if diff == 0:
                 continue
             changed = True
+            scale_types[container_type] = requested
             while diff < 0:
                 c = containers.pop()
                 to_remove.append(c)
@@ -267,11 +292,15 @@ def scale(self, user, structure):  # noqa
                 to_add.append(c)
                 container_num += 1
                 diff -= 1
+
         if changed:
-            if to_add:
-                self._start_containers(to_add)
-            if to_remove:
-                self._destroy_containers(to_remove)
+            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)
         # save new structure to the database
         vals = self.container_set.exclude(type='run').values(
             'type').annotate(Count('pk')).order_by()
@@ -281,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:
@@ -334,21 +388,50 @@ def deploy(self, user, release):
         """Deploy a new release to this application"""
         existing = self.container_set.exclude(type='run')
         new = []
+        scale_types = set()
         for e in existing:
             n = e.clone(release)
             n.save()
             new.append(n)
+            scale_types.add(e.type)
 
-        self._start_containers(new)
+        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