Merge pull request #4101 from mboersma/scheduler-superclass

ref(controller): share common scheduler code

Author: mboersma

controller/scheduler/__init__.py

@@ -0,0 +1,35 @@
+
+class AbstractSchedulerClient(object):
+    """
+    A generic interface to a scheduler backend.
+    """
+
+    def __init__(self, target, auth, options, pkey):
+        self.target = target
+        self.auth = auth
+        self.options = options
+        self.pkey = pkey
+
+    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

controller/scheduler/chaos.py

@@ -1,4 +1,5 @@
 import random
+
 from .mock import MockSchedulerClient, jobs
 from .states import JobState
 
@@ -12,52 +13,38 @@
 class ChaosSchedulerClient(MockSchedulerClient):
 
     def create(self, name, image, command, **kwargs):
+        """Create a new container."""
         if random.random() < CREATE_ERROR_RATE:
-            job = jobs.get(name, {})
-            job.update({'state': JobState.error})
-            jobs[name] = job
-            return
-        return super(ChaosSchedulerClient, self).create(name, image, command, **kwargs)
+            jobs.setdefault(name, {})['state'] = JobState.error
+        else:
+            super(ChaosSchedulerClient, self).create(name, image, command, **kwargs)
 
     def destroy(self, name):
-        """
-        Destroy an existing job
-        """
+        """Destroy a container."""
         if random.random() < DESTROY_ERROR_RATE:
-            job = jobs.get(name, {})
-            job.update({'state': JobState.error})
-            jobs[name] = job
-            return
-        return super(ChaosSchedulerClient, self).destroy(name)
+            jobs.setdefault(name, {})['state'] = JobState.error
+        else:
+            super(ChaosSchedulerClient, self).destroy(name)
 
     def run(self, name, image, entrypoint, command):
-        """
-        Run a one-off command
-        """
+        """Run a one-off command."""
         if random.random() < CREATE_ERROR_RATE:
             raise RuntimeError('exit code 1')
-        return super(ChaosSchedulerClient, self).run(name, image, entrypoint, command)
+        else:
+            super(ChaosSchedulerClient, self).run(name, image, entrypoint, command)
 
     def start(self, name):
-        """
-        Start an idle job
-        """
+        """Start a container."""
         if random.random() < START_ERROR_RATE:
-            job = jobs.get(name, {})
-            job.update({'state': JobState.crashed})
-            jobs[name] = job
-            return
-        return super(ChaosSchedulerClient, self).start(name)
+            jobs.setdefault(name, {})['state'] = JobState.crashed
+        else:
+            super(ChaosSchedulerClient, self).start(name)
 
     def stop(self, name):
-        """
-        Stop a running job
-        """
+        """Stop a container."""
         if random.random() < STOP_ERROR_RATE:
-            job = jobs.get(name, {})
-            job.update({'state': JobState.error})
-            jobs[name] = job
-            return
-        return super(ChaosSchedulerClient, self).stop(name)
+            jobs.setdefault(name, {})['state'] = JobState.crashed
+        else:
+            super(ChaosSchedulerClient, self).stop(name)
 
 SchedulerClient = ChaosSchedulerClient

controller/scheduler/fleet.py

@@ -1,15 +1,16 @@
-import cStringIO
 import base64
 import copy
+import cStringIO
 import httplib
 import json
 import paramiko
-import socket
 import re
+import socket
 import time
 
 from django.conf import settings
 
+from . import AbstractSchedulerClient
 from .states import JobState
 
 
@@ -32,13 +33,10 @@ def connect(self):
         self.sock = sock
 
 
-class FleetHTTPClient(object):
+class FleetHTTPClient(AbstractSchedulerClient):
 
     def __init__(self, target, auth, options, pkey):
-        self.target = target
-        self.auth = auth
-        self.options = options
-        self.pkey = pkey
+        super(FleetHTTPClient, self).__init__(target, auth, options, pkey)
         # single global connection
         self.conn = UHTTPConnection(self.target)
 
@@ -119,7 +117,7 @@ def _get_machines(self):
     # container api
 
     def create(self, name, image, command='', template=None, **kwargs):
-        """Create a container"""
+        """Create a container."""
         self._create_container(name, image, command,
                                template or copy.deepcopy(CONTAINER_TEMPLATE), **kwargs)
 
@@ -173,7 +171,7 @@ def _get_hostname(self, application_name):
             raise RuntimeError('Unsupported hostname: ' + hostname)
 
     def start(self, name):
-        """Start a container"""
+        """Start a container."""
         self._put_unit(name, {'desiredState': 'launched'})
         self._wait_for_container_running(name)
 
@@ -212,12 +210,12 @@ def _wait_for_destroy(self, name):
             raise RuntimeError('timeout on container destroy')
 
     def stop(self, name):
-        """Stop a container"""
+        """Stop a container."""
         self._put_unit(name, {"desiredState": "loaded"})
         self._wait_for_job_state(name, JobState.created)
 
     def destroy(self, name):
-        """Destroy a container"""
+        """Destroy a container."""
         # call all destroy functions, ignoring any errors
         try:
             self._destroy_container(name)
@@ -235,7 +233,7 @@ def _destroy_container(self, name):
                     raise
 
     def run(self, name, image, entrypoint, command):  # noqa
-        """Run a one-off command"""
+        """Run a one-off command."""
         self._create_container(name, image, command, copy.deepcopy(RUN_TEMPLATE),
                                entrypoint=entrypoint)
         # launch the container
@@ -337,13 +335,14 @@ def _do_ssh(cmd):
         return rc, output
 
     def state(self, name):
+        """Display the given job's running state."""
         systemdActiveStateMap = {
-            "active": "up",
-            "reloading": "down",
-            "inactive": "created",
-            "failed": "crashed",
-            "activating": "down",
-            "deactivating": "down",
+            'active': 'up',
+            'reloading': 'down',
+            'inactive': 'created',
+            'failed': 'crashed',
+            'activating': 'down',
+            'deactivating': 'down',
         }
         try:
             # NOTE (bacongobbler): this call to ._get_unit() acts as a pre-emptive check to
@@ -354,9 +353,8 @@ def state(self, name):
             # 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.
-            if activeState == 'failed':
-                if state['systemdLoadState'] == 'loaded':
-                    return JobState.created
+            if activeState == 'failed' and state['systemdLoadState'] == 'loaded':
+                return JobState.created
             return getattr(JobState, systemdActiveStateMap[activeState])
         except KeyError:
             # failed retrieving a proper response from the fleet API
@@ -366,12 +364,6 @@ def state(self, name):
             # which means it does not exist
             return JobState.destroyed
 
-    def attach(self, name):
-        """
-        Attach to a job's stdin, stdout and stderr
-        """
-        raise NotImplementedError
-
 SchedulerClient = FleetHTTPClient
 
 

controller/scheduler/k8s.py

@@ -1,13 +1,16 @@
 import copy
-import json
 import httplib
-import time
+import json
+import random
 import re
 import string
+import time
+
 from django.conf import settings
-from .states import JobState
 from docker import Client
-import random
+from .states import JobState
+from . import AbstractSchedulerClient
+
 
 POD_TEMPLATE = '''{
   "kind": "Pod",
@@ -95,9 +98,10 @@
     r'(?P<app>[a-z0-9-]+)_?(?P<version>v[0-9]+)?\.?(?P<c_type>[a-z-_]+)')
 
 
-class KubeHTTPClient():
+class KubeHTTPClient(AbstractSchedulerClient):
 
     def __init__(self, target, auth, options, pkey):
+        super(KubeHTTPClient, self).__init__(target, auth, options, pkey)
         self.target = settings.K8S_MASTER
         self.port = "8080"
         self.registry = settings.REGISTRY_HOST+":"+settings.REGISTRY_PORT
@@ -343,6 +347,7 @@ def _create_rc(self, name, image, command, **kwargs):
         return json.loads(data)
 
     def create(self, name, image, command, **kwargs):
+        """Create a container."""
         self._create_rc(name, image, command, **kwargs)
         app_type = name.split(".")[1]
         name = name.replace(".", "-")
@@ -433,16 +438,12 @@ def _create_service(self, name, app_name, app_type):
             raise RuntimeError(errmsg)
 
     def start(self, name):
-        """
-        Start a container
-        """
-        return
+        """Start a container."""
+        pass
 
     def stop(self, name):
-        """
-        Stop a container
-        """
-        return
+        """Stop a container."""
+        pass
 
     def _delete_rc(self, name):
         headers = {'Content-Type': 'application/json'}
@@ -460,9 +461,7 @@ def _delete_rc(self, name):
             raise RuntimeError(errmsg)
 
     def destroy(self, name):
-        """
-        Destroy a the app
-        """
+        """Destroy a container."""
         name = name.split(".")
         name = name[0]+'-'+name[1]
         name = name.replace("_", "-")
@@ -586,9 +585,7 @@ def logs(self, name):
         return log_data
 
     def run(self, name, image, entrypoint, command):
-        """
-        Run a one-off command
-        """
+        """Run a one-off command."""
         name = name.replace(".", "-")
         name = name.replace("_", "-")
         l = {}
@@ -666,17 +663,12 @@ def _get_pod_state(self, name):
             return JobState.destroyed
 
     def state(self, name):
+        """Display the given job's running state."""
         try:
             return self._get_pod_state(name)
         except KeyError:
             return JobState.error
         except RuntimeError:
             return JobState.destroyed
 
-    def attach(self, name):
-        """
-        Attach to a job's stdin, stdout and stderr
-        """
-        return NotImplementedError
-
 SchedulerClient = KubeHTTPClient