feat(swarm): add run functionality

feat(swarm): add default value to swarm host key in templates

feat(swarm): add deis/scheduler keys to confd

fix(swarm): adjust swarm.py python coding style

fix(swarm): add confd template

fix(swarm): modify swarm host etcd key

fix(swarm): change swarm etcdpath end point

fix(scheduler): create seperate etcd directory for deis scheduler

fix(swarm): make docker-py client cpu limits compatible with docker swarm

fix(tests): check welcome to ngnix page if error is 503

Author: Sivaram Mothiki

conf.d/confd_settings.toml

@@ -10,6 +10,6 @@ keys = [
   "/deis/registry",
   "/deis/domains",
   "/deis/platform",
-  "/deis/scheduler/swarm/host",
+  "/deis/scheduler",
 ]
 reload_cmd = "/app/bin/reload"

scheduler/swarm.py

@@ -1,31 +1,49 @@
-import logging
-from docker import Client
+import re
+import time
 
 from django.conf import settings
+from docker import Client
+
+from .states import JobState
 
 
+MATCH = re.compile(
+    r'(?P<app>[a-z0-9-]+)_?(?P<version>v[0-9]+)?\.?(?P<c_type>[a-z-_]+)?.(?P<c_num>[0-9]+)')
+
 
 class SwarmClient(object):
-    def __init__(self,target, auth, options, pkey):
-        self.target =  settings.SWARM_HOST
+
+    def __init__(self, target, auth, options, pkey):
+        self.target = settings.SWARM_HOST
         # single global connection
-        self.registry = settings.REGISTRY_HOST+":"+settings.REGISTRY_PORT
-        self.docker_cli = Client(base_url='tcp://'+self.target+':'+"2395",timeout=1200)
+        self.registry = settings.REGISTRY_HOST + ':' + settings.REGISTRY_PORT
+        self.docker_cli = Client("tcp://{}:2395".format(self.target),
+                                 timeout=1200, version='1.17')
 
     def create(self, name, image, command='', template=None, **kwargs):
         """Create a container"""
-        cimage=self.registry+"/"+image
-        cname=name
-        ccommand=command
-        # self.docker_cli.pull(cimage, stream=False,insecure_registry=True)
-        self.docker_cli.create_container(image=cimage,name=cname,command=ccommand)#,hostname=self._get_hostname(cname),ports=self._get_ports(cimage))
-        self.docker_cli.start(cname, port_bindings=self._get_portbindings(cimage),publish_all_ports=True)
+        cimage = self.registry + '/' + image
+        affinity = "affinity:container!=~/{}*/".format(re.split(r'_v\d.', name)[0])
+        l = locals().copy()
+        l.update(re.match(MATCH, name).groupdict())
+        mem = kwargs.get('memory', {}).get(l['c_type'])
+        if mem:
+            mem = mem.lower()
+            if mem[-2:-1].isalpha() and mem[-1].isalpha():
+                mem = mem[:-1]
+        cpu = kwargs.get('cpu', {}).get(l['c_type'])
+        self.docker_cli.create_container(image=cimage, name=name,
+                                         command=command.encode('utf-8'), mem_limit=mem,
+                                         cpu_shares=cpu,
+                                         environment=[affinity])
+        self.docker_cli.stop(name)
 
     def start(self, name):
         """
         Start a container
         """
-        self.docker_cli.start(name)
+        self.docker_cli.start(name, publish_all_ports=True)
+
         return
 
     def stop(self, name):
@@ -34,54 +52,94 @@ def stop(self, name):
         """
         self.docker_cli.stop(name)
         return
+
     def destroy(self, name):
         """
         Destroy a container
         """
-        self.docker_cli.stop(name)
+        self.stop(name)
         self.docker_cli.remove_container(name)
         return
 
     def run(self, name, image, entrypoint, command):
         """
         Run a one-off command
         """
-        # dump input into a json object for testing purposes
-        return 0, json.dumps({'name': name,
-                              'image': image,
-                              'entrypoint': entrypoint,
-                              'command': command})
+        cimage = self.registry + '/' + image
+        # use affinity for nodes that already have the image
+        affinity = "affinity:image==~{}".format(cimage)
+        self.docker_cli.create_container(image=cimage, name=name,
+                                         command=command.encode('utf-8'),
+                                         environment=[affinity],
+                                         entrypoint=[entrypoint])
+        time.sleep(2)
+        self.start(name)
+        rc = 0
+        while (True):
+            if self._get_container_state(name) == JobState.created:
+                break
+            time.sleep(1)
+        try:
+            output = self.docker_cli.logs(name)
+            return rc, output
+        except:
+            rc = 1
+            return rc, output
+
+    def _get_container_state(self, name):
+        try:
+            if self.docker_cli.inspect_container(name)['State']['Running']:
+                return JobState.up
+            else:
+                return JobState.created
+        except:
+            return JobState.destroyed
+
+    def state(self, name):
+        try:
+            # NOTE (bacongobbler): this call to ._get_unit() acts as a pre-emptive check to
+            # determine if the job no longer exists (will raise a RuntimeError on 404)
+            for _ in range(30):
+                return self._get_container_state(name)
+                time.sleep(1)
+            # FIXME (bacongobbler): when fleet loads a job, sometimes it'll automatically start and
+            # stop the container, which in our case will return as 'failed', even though
+            # the container is perfectly fine.
+        except KeyError:
+            # failed retrieving a proper response from the fleet API
+            return JobState.error
+        except RuntimeError:
+            # failed to retrieve a response from the fleet API,
+            # which means it does not exist
+            return JobState.destroyed
 
     def attach(self, name):
         """
         Attach to a job's stdin, stdout and stderr
         """
-        return StringIO(), StringIO(), StringIO()
+        raise NotImplementedError
 
     def _get_hostname(self, application_name):
         hostname = settings.UNIT_HOSTNAME
-        if hostname == "default":
+        if hostname == 'default':
             return ''
-        elif hostname == "application":
+        elif hostname == 'application':
             # replace underscore with dots, since underscore is not valid in DNS hostnames
-            dns_name = application_name.replace("_", ".")
+            dns_name = application_name.replace('_', '.')
             return dns_name
-        elif hostname == "server":
+        elif hostname == 'server':
             raise NotImplementedError
         else:
             raise RuntimeError('Unsupported hostname: ' + hostname)
 
-    def _get_portbindings(self,image):
-        dictports=self.docker_cli.inspect_image(image)["ContainerConfig"]["ExposedPorts"]
-        for port,mapping in dictports.items():
-            dictports[port]=None
+    def _get_portbindings(self, image):
+        dictports = self.docker_cli.inspect_image(image)['ContainerConfig']['ExposedPorts']
+        for port, mapping in dictports.items():
+            dictports[port] = None
         return dictports
 
-    def _get_ports(self,image):
-        ports=[]
-        dictports=self.docker_cli.inspect_image(image)["ContainerConfig"]["ExposedPorts"]
-        for port,mapping in dictports.items():
-            ports.append(int(port.split('/')[0]))
-        return ports
+    def _get_ports(self, image):
+        dictports = self.docker_cli.inspect_image(image)['ContainerConfig']['ExposedPorts']
+        return [int(port.split('/')[0]) for port in dictports.iterkeys()]
 
 SchedulerClient = SwarmClient

templates/confd_settings.py

@@ -11,7 +11,9 @@
     SCHEDULER_OPTIONS = {}
 
 # scheduler swarm manager host
-SWARM_HOST = """{{ or (.deis_scheduler_swarm_host) "" }}"""
+SWARM_HOST = '{{ if exists "/deis/scheduler/swarm/host" }}{{ getv "/deis/scheduler/swarm/host" }}{{ else }}127.0.0.1{{ end }}'
+#SWARM_HOST = '{{ or (.deis_scheduler_swarm_host) "127.0.0.1" }}'
+
 
 # base64-encoded SSH private key to facilitate current version of "deis run"
 SSH_PRIVATE_KEY = """{{ if exists "/deis/platform/sshPrivateKey" }}{{ getv "/deis/platform/sshPrivateKey" }}{{ else }}""{{end}}"""