refactor(scheduler): coreos scheduler implementation

Note an SSH wrapper is used to get around fleetctl ssh-agent requirement.  We will ultimately want to transition to an HTTP API for talking to fleet.  Until then, we should explore modifying fleetctl to remove the ssh-agent requirement for tunneling.

add cluster setup and teardown

announce to /deis/services/{container_name} with a sidekick container

 - add pull before start
 - add stop and remove in ExecStop
 - use hardcoded expose port 5000 for deis/slugrunner

parallelize scheduler tasks, add appropriate blocking

exclude scheduler from max-line-length

add sidekick container for logging, unit cleanup

add unit name regex, use it when templating units

add domain and options to SchedulerClient
fix destroy bug

update readme to reflect new scheduler workflow, with placeholders for not-yet-implemented features

first pass at systemd units for components

coreos contrib scripts, units and user-data helpers

Author: Gabriel Monroy

Dockerfile

@@ -15,6 +15,9 @@ RUN wget -qO- https://raw.github.com/pypa/pip/1.5.4/contrib/get-pip.py | python
 # install requirements before ADD to cache layer and speed build
 RUN pip install boto==2.23.0 celery==3.1.8 Django==1.6.2 django-allauth==0.15.0 django-guardian==1.1.1 django-json-field==0.5.5 django-yamlfield==0.5 djangorestframework==2.3.12 dop==0.1.4 gevent==1.0 gunicorn==18.0 paramiko==1.12.1 psycopg2==2.5.2 pycrypto==2.6.1 python-etcd==0.3.0 pyrax==1.6.2 PyYAML==3.10 redis==2.8.0 static==1.0.2 South==0.8.4
 
+# install openssh-client for temporary fleetctl wrapper
+RUN apt-get install -yq openssh-client
+
 # clone the project into /app
 ADD . /app
 

api/models.py

@@ -60,7 +60,8 @@ class Cluster(UuidAuditedModel):
     Cluster used to run jobs
     """
 
-    CLUSTER_TYPES = (('mock', 'Mock Cluster'),)
+    CLUSTER_TYPES = (('mock', 'Mock Cluster'),
+                     ('coreos', 'CoreOS Cluster'))
 
     owner = models.ForeignKey(settings.AUTH_USER_MODEL)
     id = models.CharField(max_length=128, unique=True)
@@ -74,6 +75,26 @@ class Cluster(UuidAuditedModel):
     def __str__(self):
         return self.id
 
+    def _get_scheduler(self, *args, **kwargs):
+        module_name = 'scheduler.' + self.type
+        mod = importlib.import_module(module_name)
+        return mod.SchedulerClient(self.id, self.hosts, self.auth,
+                                   self.domain, self.options)
+
+    _scheduler = property(_get_scheduler)
+
+    def create(self):
+        """
+        Initialize a cluster's router and log aggregator
+        """
+        return tasks.create_cluster.delay(self).get()
+
+    def destroy(self):
+        """
+        Destroy a cluster's router and log aggregator
+        """
+        return tasks.destroy_cluster.delay(self).get()
+
 
 @python_2_unicode_compatible
 class App(UuidAuditedModel):
@@ -92,18 +113,17 @@ class Meta:
     def __str__(self):
         return self.id
 
-    def init(self, *args, **kwargs):
+    def create(self, *args, **kwargs):
         config = Config.objects.create(owner=self.owner, app=self, values={})
         build = Build.objects.create(owner=self.owner, app=self, image=settings.DEFAULT_BUILD)
         Release.objects.create(version=1, owner=self.owner, app=self, config=config, build=build)
 
-    def delete(self, *args, **kwargs):
+    def destroy(self, *args, **kwargs):
         for c in self.container_set.all():
             c.destroy()
-        super(App, self).delete(*args, **kwargs)
 
     def deploy(self, release):
-        return tasks.deploy_release.delay(self, release)
+        return tasks.deploy_release.delay(self, release).get()
 
     def scale(self, **kwargs):
         """Scale containers up or down to match requested."""
@@ -200,67 +220,67 @@ class Meta:
         get_latest_by = '-created'
         ordering = ['created']
 
-    def _scheduler(self, *args, **kwargs):
-        module_name = 'scheduler.' + self.app.cluster.type
-        mod = importlib.import_module(module_name)
-        return mod.SchedulerClient(self.app.cluster.id,
-                                   self.app.cluster.hosts,
-                                   self.app.cluster.auth)
-
-    def job_id(self):
+    def _get_job_id(self):
         app = self.app.id
         release = self.release
         version = "v{}".format(release.version)
         num = self.num
         c_type = self.type
         if not c_type:
-            job_id = "{app}.{version}.{num}".format(**locals())
+            job_id = "{app}_{version}.{num}".format(**locals())
         else:
-            job_id = "{app}_{c_type}.{version}.{num}".format(**locals())
+            job_id = "{app}_{version}.{c_type}.{num}".format(**locals())
         return job_id
 
+    _job_id = property(_get_job_id)
+
+    def _get_scheduler(self):
+        return self.app.cluster._scheduler
+
+    _scheduler = property(_get_scheduler)
+
     def create(self):
         image = self.release.build.image
-        self._scheduler().create(self.job_id(), image, 'docker run {image}'.format(**locals()))
+        self._scheduler.create(self._job_id, image, 'docker run {image}'.format(**locals()))
         self.state = 'created'
         self.save()
 
     def start(self):
         self.state = 'starting'
         self.save()
-        self._scheduler().start(self.job_id())
+        self._scheduler.start(self._job_id)
         self.state = 'up'
         self.save()
 
     def deploy(self, release):
-        old_job_id = self.job_id()
+        old_job_id = self._job_id
         self.state = 'deploying'
         self.save()
         # update release
         self.release = release
         self.save()
         # deploy new container
-        new_job_id = self.job_id()
+        new_job_id = self._job_id
         image = self.release.build.image
-        self._scheduler().create(new_job_id, image, 'docker run {image}'.format(**locals()))
-        self._scheduler().start(new_job_id)
+        self._scheduler.create(new_job_id, image, 'docker run {image}'.format(**locals()))
+        self._scheduler.start(new_job_id)
         # destroy old container
-        self._scheduler().destroy(old_job_id)
+        self._scheduler.destroy(old_job_id)
         self.state = 'up'
         self.save()
 
     def stop(self):
         self.state = 'stopping'
         self.save()
-        self._scheduler().stop(self.job_id())
+        self._scheduler.stop(self._job_id)
         self.state = 'stopped'
         self.save()
 
     def destroy(self):
         self.state = 'destroying'
         self.save()
         # TODO: add check for active connections before killing
-        self._scheduler().destroy(self.job_id())
+        self._scheduler.destroy(self._job_id)
         self.state = 'destroyed'
         self.save()
 

api/tasks.py

@@ -7,43 +7,72 @@
 
 from __future__ import unicode_literals
 
+import threading
+
 from celery import task
 
 
+@task
+def create_cluster(cluster):
+    cluster._scheduler.setUp()
+
+
+@task
+def destroy_cluster(cluster):
+    for app in cluster.app_set.all():
+        app.destroy()
+    cluster._scheduler.tearDown()
+
+
 @task
 def deploy_release(app, release):
     containers = app.container_set.all()
-    # TODO: parallelize
+    threads = []
     for c in containers:
-        try:
-            c.deploy(release)
-        except Exception:
+        threads.append(threading.Thread(target=c.deploy, args=(release,)))
+    try:
+        [t.start() for t in threads]
+        [t.join() for t in threads]
+    except Exception:
+        for c in containers:
             c.state = 'error'
             c.save()
-            raise
+        raise
 
 
 @task
 def start_containers(containers):
-    # TODO: parallelize
+    create_threads = []
+    start_threads = []
     for c in containers:
-        try:
-            c.create()
-            c.start()
-        except Exception:
+        create_threads.append(threading.Thread(target=c.create))
+        start_threads.append(threading.Thread(target=c.start))
+    try:
+        [t.start() for t in create_threads]
+        [t.join() for t in create_threads]
+        [t.start() for t in start_threads]
+        [t.join() for t in start_threads]
+    except Exception:
+        for c in containers:
             c.state = 'error'
             c.save()
             raise
 
 
 @task
 def stop_containers(containers):
-    # TODO: parallelize
+    destroy_threads = []
+    delete_threads = []
     for c in containers:
-        try:
-            c.destroy()
-            c.delete()
-        except Exception:
+        destroy_threads.append(threading.Thread(target=c.destroy))
+        delete_threads.append(threading.Thread(target=c.delete))
+    try:
+        [t.start() for t in destroy_threads]
+        [t.join() for t in destroy_threads]
+        [t.start() for t in delete_threads]
+        [t.join() for t in delete_threads]
+    except Exception:
+        for c in containers:
             c.state = 'error'
             c.save()
             raise

api/tests/test_container.py

@@ -9,14 +9,14 @@
 import json
 
 from django.contrib.auth.models import User
-from django.test import TestCase
+from django.test import TransactionTestCase
 from django.test.utils import override_settings
 
 from api.models import Container, App
 
 
 @override_settings(CELERY_ALWAYS_EAGER=True)
-class ContainerTest(TestCase):
+class ContainerTest(TransactionTestCase):
 
     """Tests creation of containers on nodes"""
 

api/views.py

@@ -194,6 +194,13 @@ def pre_save(self, obj):
         if not hasattr(obj, 'owner'):
             obj.owner = self.request.user
 
+    def post_save(self, cluster, created=False, **kwargs):
+        if created:
+            cluster.create()
+
+    def pre_delete(self, cluster):
+        cluster.destroy()
+
 
 class AppPermsViewSet(viewsets.ViewSet):
     """RESTful views for sharing apps with collaborators."""
@@ -272,7 +279,7 @@ def get_queryset(self, **kwargs):
 
     def post_save(self, app, created=False, **kwargs):
         if created:
-            app.init()
+            app.create()
 
     def scale(self, request, **kwargs):
         new_structure = {}