ref(scheduler): split up k8s.py and make it more modular

Splits each resource type into its own class, creates a KubeHTTPClient that handles HTTP interactions
and moves the k8s code up into __init__. Separation of responsibility in the code is stronger as the
scheduler code the Deis API interfaces with doesn't touch HTTP and individual resource classes deal
very little with it as well.

Many HTTP interactions have been consolidated and by default return a RunTimeError Exception if a
non-2xx code is returned unless the check=False flag is passed

Author: helgi

rootfs/api/models.py

@@ -164,9 +164,7 @@ class Meta:
     @property
     def _scheduler(self):
         mod = importlib.import_module(settings.SCHEDULER_MODULE)
-        return mod.SchedulerClient(settings.SCHEDULER_URL,
-                                   settings.SCHEDULER_AUTH,
-                                   settings.SCHEDULER_OPTIONS)
+        return mod.SchedulerClient()
 
     def __str__(self):
         return self.id

rootfs/scheduler/__init__.py

@@ -1,34 +1,149 @@
+import logging
 
-class AbstractSchedulerClient(object):
-    """
-    A generic interface to a scheduler backend.
-    """
+from .states import JobState
+from .abstract import AbstractSchedulerClient
+from .client import unhealthy
+from .replicationcontroller import ReplicationController
+from .pod import Pod
+from .service import Service
+from .namespace import Namespace
 
-    def __init__(self, target, auth, options):
-        self.target = target
-        self.auth = auth
-        self.options = options
+log = logging.getLogger(__name__)
+
+
+class KubeSchedulerClient(AbstractSchedulerClient):
+    def deploy(self, name, image, command, **kwargs):
+        log.debug('deploy %s, img %s, params %s, cmd "%s"', name, image, kwargs, command)
+        app_name = kwargs.get('aname', {})
+        app_type = name.split('.')[1]
+
+        controller = ReplicationController()
+        old_rc = controller.old(app_name, app_type)
+        new_rc = controller.create(name, image, command, **kwargs)
+        if old_rc:
+            desired = int(old_rc['spec']['replicas'])
+            old_rc_name = old_rc['metadata']['name']
+        else:
+            desired = 1
+
+        new_rc_name = new_rc['metadata']['name']
+        try:
+            count = 1
+            while desired >= count:
+                new_rc = controller.scale(new_rc_name, count, app_name)
+                if old_rc:
+                    old_rc = controller.scale(old_rc_name, desired-count, app_name)
+                count += 1
+        except Exception as e:
+            controller.scale(new_rc['metadata']['name'], 0, app_name)
+            controller.delete(new_rc['metadata']['name'], app_name)
+            if old_rc:
+                controller.scale(old_rc['metadata']['name'], desired, app_name)
+
+            raise RuntimeError('{} (deploy): {}'.format(name, e))
+
+        if old_rc:
+            controller.delete(old_rc_name, app_name)
+
+    def scale(self, name, image, command, **kwargs):
+        log.debug('scale %s, img %s, params %s, cmd "%s"', name, image, kwargs, command)
+        app_name = kwargs.get('aname', {})
+        rc_name = name.replace('.', '-').replace('_', '-')
+
+        controller = ReplicationController()
+        # Create if ReplicationController doesn't exist
+        if unhealthy(controller.status(rc_name, app_name)):
+            self.create(name, image, command, **kwargs)
+            return
+
+        name = name.replace('.', '-').replace('_', '-')
+        num = kwargs.get('num', {})
+        old_replicas = controller.get(name, app_name).json()['spec']['replicas']
+        try:
+            controller.scale(name, num, app_name)
+        except Exception as e:
+            controller.scale(name, old_replicas, app_name)
+            raise RuntimeError('{} (Scale): {}'.format(name, e))
 
     def create(self, name, image, command, **kwargs):
-        """Create a new container."""
-        raise NotImplementedError
+        """Create a container."""
+        log.debug('create %s, img %s, params %s, cmd "%s"', name, image, kwargs, command)
+        controller = ReplicationController()
+        controller.create(name, image, command, **kwargs)
+        app_type = name.split('.')[1]
+        name = name.replace('.', '-').replace('_', '-')
+        app_name = kwargs.get('aname', {})
+        try:
+            Service().create(name, app_name, app_type)
+        except:
+            controller.scale(name, 0, app_name)
+            controller.delete(name, app_name)
+            raise
+
+    def start(self, name):
+        """Start a container."""
+        pass
+
+    def stop(self, name):
+        """Stop a container."""
+        pass
 
     def destroy(self, name):
-        """Destroy a container."""
-        raise NotImplementedError
+        """Destroy a application by deleting its namespace."""
+        log.debug('destroy %s', name)
+        namespace = Namespace()
+        response = namespace.get(name, check=False)
+        if response.status_code == 404:
+            log.warn('delete Namespace "%s": not found', name)
+        else:
+            namespace.delete(name)
+
+    def logs(self, name):
+        """Aggregate logs from all pods in the namespace"""
+        log.debug("logs %s", name)
+        app_name = name.split('_')[0]
+        name = name.replace('.', '-').replace('_', '-')
+
+        pod = Pod()
+        items = pod.all(app_name).json()['items']
+        log_data = ''
+        for data in items:
+            if name in data['metadata']['generateName'] and data['status']['phase'] == 'Running':
+                log_data += pod.log(pod['metadata']['name'], app_name).text
+
+        return log_data
 
     def run(self, name, image, entrypoint, command):
         """Run a one-off command."""
-        raise NotImplementedError
-
-    def start(self, name):
-        """Start a container."""
-        raise NotImplementedError
+        log.debug('run %s, img %s, entypoint %s, cmd "%s"', name, image, entrypoint, command)
+        return Pod().run(name, image, entrypoint, command)
 
     def state(self, name):
-        """Display the given job's running state."""
-        raise NotImplementedError
+        """Display the state of a container."""
+        # See "Pod Phase" at http://kubernetes.io/v1.1/docs/user-guide/pod-states.html
+        phase_states = {
+            'Pending': JobState.initialized,
+            'Running': JobState.up,
+            'Succeeded': JobState.down,
+            'Failed': JobState.crashed,
+            'Unknown': JobState.error,
+        }
 
-    def stop(self, name):
-        """Stop a container."""
-        raise NotImplementedError
+        try:
+            app_name = name.split('_')[0]
+            name = name.split('.')
+            name = name[0] + '-' + name[1]
+            name = name.replace('_', '-')
+
+            items = Pod().all(app_name).json()['items']
+            for data in items:
+                if data['metadata']['generateName'] == name + '-':
+                    phase = data['status']['phase']
+                    return phase_states[phase]
+
+            return JobState.destroyed
+        except Exception as err:
+            log.warn(err)
+            return JobState.error
+
+SchedulerClient = KubeSchedulerClient

rootfs/scheduler/abstract.py

@@ -0,0 +1,29 @@
+
+class AbstractSchedulerClient(object):
+    """
+    A generic interface to a scheduler backend.
+    """
+
+    def create(self, name, image, command, **kwargs):
+        """Create a new container."""
+        raise NotImplementedError
+
+    def destroy(self, name):
+        """Destroy a container."""
+        raise NotImplementedError
+
+    def run(self, name, image, entrypoint, command):
+        """Run a one-off command."""
+        raise NotImplementedError
+
+    def start(self, name):
+        """Start a container."""
+        raise NotImplementedError
+
+    def state(self, name):
+        """Display the given job's running state."""
+        raise NotImplementedError
+
+    def stop(self, name):
+        """Stop a container."""
+        raise NotImplementedError

rootfs/scheduler/base.py

@@ -0,0 +1,8 @@
+from django.conf import settings
+from .client import KubeHTTPClient
+
+
+class Base(object):
+    def __init__(self):
+        self.registry = settings.REGISTRY_URL
+        self.api = KubeHTTPClient(settings.SCHEDULER_URL)

rootfs/scheduler/client.py

@@ -0,0 +1,130 @@
+import requests
+import urlparse
+
+
+def error(response, errmsg, *args):
+    # reversing since URL vs Messages are constructed the opposite way
+    args = tuple(reversed(args))
+    errmsg = errmsg.format(*args)
+    errmsg = "failed to {}: {} {}\n{}".format(
+        errmsg,
+        response.status_code,
+        response.reason,
+        response.json()
+    )
+    raise RuntimeError(errmsg)
+
+
+def error_message(method, url):
+    """ Get good error message based on URL and method used """
+    errmsg = method
+    count = url.count('{}')
+    segments = url.strip('/').split('/')
+
+    # Format is /namespace/{}/<resource_type>/{}/<sub_resource>
+    # Determine if resource type should be plural or singular
+    resource_type = segments[3][0:-1] if count > 1 else segments[3]
+    # Check if it is a sub resource
+    if len(segments) > 4:
+        errmsg += ' %s for' % segments.pop()
+
+    errmsg += ' %s "{}"' % resource_type.capitalize()
+
+    # Requesting a namespaced resource
+    if count > 1:
+        errmsg += ' in Namespace "{}"'
+
+    return errmsg
+
+
+def unhealthy(status_code):
+    """Status is considered unhealthy if it is not 2xx"""
+    if not 200 <= status_code <= 299:
+        return True
+
+    return False
+
+
+class KubeHTTPClient(object):
+    version = 'v1'
+    session = None
+
+    def __init__(self, target):
+        self.target = target
+
+        # TODO singleton this if it makes sense
+        if not self.session:
+            self.connect()
+
+    def connect(self):
+        with open('/var/run/secrets/kubernetes.io/serviceaccount/token') as token_file:
+            token = token_file.read()
+
+        session = requests.Session()
+        session.headers = {
+            'Authorization': 'Bearer ' + token,
+            'Content-Type': 'application/json',
+        }
+        session.verify = '/var/run/secrets/kubernetes.io/serviceaccount/ca.crt'
+        self.session = session
+
+    def url(self, tmpl, *args):
+        """Return a fully-qualified Kubernetes API URL from a string template with args."""
+        url = "/api/{}".format(self.version) + tmpl.format(*args)
+        return urlparse.urljoin(self.target, url)
+
+    def get(self, url, *args, **kwargs):
+        response = self.session.get(self.url(url, *args), **kwargs)
+        check = kwargs.get('check', True)  # Check response and fail
+        if check and unhealthy(response.status_code):
+            error(response, error_message('get', url), *args)
+
+        return response
+
+    def delete(self, url, *args, **kwargs):
+        response = self.session.delete(self.url(url, *args), **kwargs)
+        check = kwargs.get('check', True)  # Check response and fail
+        if check and unhealthy(response.status_code):
+            error(response, error_message('delete', url), *args)
+
+        return response
+
+    def post(self, url, *args, **kwargs):
+        response = self.session.post(self.url(url, *args), **kwargs)
+        check = kwargs.get('check', True)  # Check response and fail
+        if check and unhealthy(response.status_code):
+            error(response, error_message('create', url), *args)
+
+        return response
+
+    def put(self, url, *args, **kwargs):
+        response = self.session.put(self.url(url, *args), **kwargs)
+        check = kwargs.get('check', True)  # Check response and fail
+        if check and unhealthy(response.status_code):
+            error(response, error_message('update', url), *args)
+
+        return response
+
+    def head(self, url, check=True, *args, **kwargs):
+        response = self.session.head(self.url(url, *args), **kwargs)
+        check = kwargs.get('check', True)  # Check response and fail
+        if check and unhealthy(response.status_code):
+            error(response, error_message('head', url), *args)
+
+        return response
+
+    def options(self, url, *args, **kwargs):
+        response = self.session.head(self.url(url, *args), **kwargs)
+        check = kwargs.get('check', True)  # Check response and fail
+        if check and unhealthy(response.status_code):
+            error(response, error_message('options', url), *args)
+
+        return response
+
+    def patch(self, url, *args, **kwargs):
+        response = self.session.patch(self.url(url, *args), **kwargs)
+        check = kwargs.get('check', True)  # Check response and fail
+        if check and unhealthy(response.status_code):
+            error(response, error_message('patch', url), *args)
+
+        return response

rootfs/scheduler/events.py

@@ -0,0 +1,6 @@
+from .base import Base
+
+
+class Events(Base):
+    def get(self, namespace):
+        return self.api.get('/namespaces/{}/events', namespace)