feat(controller): add ps:restart command

Author: Matthew Fisher

client/deis.py

@@ -1700,6 +1700,7 @@ def ps(self, args):
         Valid commands for processes:
 
         ps:list        list application processes
+        ps:restart     restart process types (e.g. web, worker)
         ps:scale       scale processes (e.g. web=4 worker=2)
 
         Use `deis help [command]` to learn more.
@@ -1737,6 +1738,45 @@ def ps_list(self, args, app=None):
                 self._logger.info("{type}.{num} {state} ({release})".format(**c))
             self._logger.info('')
 
+    def ps_restart(self, args):
+        """
+        Restarts an application's processes by type.
+
+        Usage: deis ps:restart [<type>] [options]
+
+        Arguments:
+          <type>
+            the process name as defined in your Procfile, such as 'web' or 'worker'.
+
+        Options:
+          -a --app=<app>
+            the uniquely identifiable name for the application.
+        """
+        app = args.get('--app')
+        if not app:
+            app = self._session.app
+        restarting_cmd = 'Restarting processes... but first, {}!\n'.format(
+            os.environ.get('DEIS_DRINK_OF_CHOICE', 'coffee'))
+        sys.stdout.write(restarting_cmd)
+        sys.stdout.flush()
+        try:
+            progress = TextProgress()
+            progress.start()
+            before = time.time()
+            url = '/v1/apps/{}/containers/restart'.format(app)
+            if args.get('<type>'):
+                url = '/v1/apps/{}/containers/{}/restart'.format(app,
+                                                                 args.get('<type>'))
+            response = self._dispatch('post', url)
+        finally:
+            progress.cancel()
+            progress.join()
+        if response.status_code == requests.codes.ok:
+            self._logger.info('done in {}s'.format(int(time.time() - before)))
+            self.ps_list({}, app)
+        else:
+            raise ResponseError(response)
+
     def ps_scale(self, args):
         """
         Scales an application's processes by type.

controller/api/models.py

@@ -198,6 +198,13 @@ def delete(self, *args, **kwargs):
         self._clean_app_logs()
         return super(App, self).delete(*args, **kwargs)
 
+    def restart(self, **kwargs):
+        to_restart = self.container_set.all()
+        if kwargs.get('type'):
+            to_restart = self.container_set.filter(type=kwargs.get('type'))
+        self._restart_containers(to_restart)
+        return to_restart
+
     def _clean_app_logs(self):
         """Delete application logs stored by the logger component"""
         path = os.path.join(settings.DEIS_LOG_DIR, self.id + '.log')
@@ -282,6 +289,27 @@ def _start_containers(self, to_add):
             err = 'warning, some containers failed to start'
             log_event(self, err, logging.WARNING)
 
+    def _restart_containers(self, to_restart):
+        """Restarts containers via the scheduler"""
+        stop_threads = []
+        start_threads = []
+        if not to_restart:
+            # do nothing if we didn't request any containers
+            return
+        for c in to_restart:
+            stop_threads.append(threading.Thread(target=c.stop))
+            start_threads.append(threading.Thread(target=c.start))
+        [t.start() for t in stop_threads]
+        [t.join() for t in stop_threads]
+        if set([c.state for c in to_restart]) != set(['created']):
+            err = 'warning, some containers failed to stop'
+            log_event(self, err, logging.WARNING)
+        [t.start() for t in start_threads]
+        [t.join() for t in start_threads]
+        if set([c.state for c in to_restart]) != set(['up']):
+            err = 'warning, some containers failed to start'
+            log_event(self, err, logging.WARNING)
+
     def _destroy_containers(self, to_destroy):
         """Destroys containers via the scheduler"""
         destroy_threads = []

controller/api/tests/test_container.py

@@ -12,7 +12,7 @@
 
 from django.contrib.auth.models import User
 from django.test import TransactionTestCase
-from scheduler.states import TransitionNotAllowed
+from scheduler.states import TransitionError
 from rest_framework.authtoken.models import Token
 
 from api.models import App, Build, Container, Release
@@ -57,7 +57,7 @@ def test_container_state_good(self):
                                      num=1)
         self.assertEqual(c.state, 'initialized')
         # test an illegal transition
-        self.assertRaises(TransitionNotAllowed, lambda: c.start())
+        self.assertRaises(TransitionError, lambda: c.start())
         c.create()
         self.assertEqual(c.state, 'created')
         c.start()

controller/api/tests/test_scheduler.py

@@ -119,6 +119,75 @@ def test_start_chaos(self):
         states = set([c['state'] for c in response.data['results']])
         self.assertEqual(states, set(['crashed', 'up']))
 
+    def test_restart_chaos(self):
+        url = '/v1/apps'
+        response = self.client.post(url, HTTP_AUTHORIZATION='token {}'.format(self.token))
+        self.assertEqual(response.status_code, 201)
+        app_id = response.data['id']
+        # post a new build
+        url = "/v1/apps/{app_id}/builds".format(**locals())
+        body = {'image': 'autotest/example', 'sha': 'a'*40,
+                'procfile': json.dumps({'web': 'node server.js', 'worker': 'node worker.js'})}
+        response = self.client.post(url, json.dumps(body), content_type='application/json',
+                                    HTTP_AUTHORIZATION='token {}'.format(self.token))
+        self.assertEqual(response.status_code, 201)
+        url = "/v1/apps/{app_id}/containers".format(**locals())
+        response = self.client.get(url, HTTP_AUTHORIZATION='token {}'.format(self.token))
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(len(response.data['results']), 1)
+        # scale up, which will allow some crashed containers
+        url = "/v1/apps/{app_id}/scale".format(**locals())
+        body = {'web': 20, 'worker': 20}
+        response = self.client.post(url, json.dumps(body), content_type='application/json',
+                                    HTTP_AUTHORIZATION='token {}'.format(self.token))
+        self.assertEqual(response.status_code, 204)
+        # let's get chaotic
+        chaos.STOP_ERROR_RATE = 0.5
+        chaos.START_ERROR_RATE = 0.5
+        # reboot the web processes
+        url = "/v1/apps/{app_id}/containers/web/restart".format(**locals())
+        response = self.client.post(url,
+                                    content_type='application/json',
+                                    HTTP_AUTHORIZATION='token {}'.format(self.token))
+        self.assertEqual(response.status_code, 200, response.data)
+        # inspect broken containers
+        url = "/v1/apps/{app_id}/containers".format(**locals())
+        response = self.client.get(url, HTTP_AUTHORIZATION='token {}'.format(self.token))
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.data['count'], 40)
+        # make sure some failed
+        states = set([c['state'] for c in response.data['results']])
+        self.assertEqual(states, set(['crashed', 'up']))
+        # make sure that we only rebooted the web processes
+        types = set([c['type'] for c in response.data['results'] if c['state'] == 'crashed'])
+        self.assertEqual(types, set(['web']))
+        # start fresh
+        chaos.STOP_ERROR_RATE = 0.0
+        chaos.START_ERROR_RATE = 0.0
+        url = "/v1/apps/{app_id}/containers/web/restart".format(**locals())
+        response = self.client.post(url,
+                                    content_type='application/json',
+                                    HTTP_AUTHORIZATION='token {}'.format(self.token))
+        # let the carnage continue
+        chaos.STOP_ERROR_RATE = 0.5
+        chaos.START_ERROR_RATE = 0.5
+        # reboot ALL the containers!
+        url = "/v1/apps/{app_id}/containers/restart".format(**locals())
+        response = self.client.post(url,
+                                    content_type='application/json',
+                                    HTTP_AUTHORIZATION='token {}'.format(self.token))
+        self.assertEqual(response.status_code, 200)
+        # inspect broken containers
+        url = "/v1/apps/{app_id}/containers".format(**locals())
+        response = self.client.get(url, HTTP_AUTHORIZATION='token {}'.format(self.token))
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(len(response.data['results']), 40)
+        # make sure some failed
+        states = set([c['state'] for c in response.data['results']])
+        self.assertEqual(states, set(['crashed', 'up']))
+        types = set([c['type'] for c in response.data['results']])
+        self.assertEqual(types, set(['web', 'worker']))
+
     def test_destroy_chaos(self):
         url = '/v1/apps'
         response = self.client.post(url, HTTP_AUTHORIZATION='token {}'.format(self.token))

controller/api/urls.py

@@ -26,6 +26,10 @@
     url(r'^apps/(?P<id>{})/releases/?'.format(settings.APP_URL_REGEX),
         views.ReleaseViewSet.as_view({'get': 'list'})),
     # application infrastructure
+    url(r'^apps/(?P<id>{})/containers/restart/?'.format(settings.APP_URL_REGEX),
+        views.ContainerViewSet.as_view({'post': 'restart'})),
+    url(r'^apps/(?P<id>{})/containers/(?P<type>[-_\w.]+)/restart/?'.format(settings.APP_URL_REGEX),
+        views.ContainerViewSet.as_view({'post': 'restart'})),
     url(r'^apps/(?P<id>{})/containers/(?P<type>[-_\w]+)/(?P<num>[-_\w]+)/?'.format(
         settings.APP_URL_REGEX),
         views.ContainerViewSet.as_view({'get': 'retrieve'})),

controller/api/views.py

@@ -213,6 +213,14 @@ def get_object(self, **kwargs):
         qs = self.get_queryset(**kwargs)
         return qs.get(num=self.kwargs['num'])
 
+    def restart(self, *args, **kwargs):
+        try:
+            containers = self.get_app().restart(**kwargs)
+            serializer = self.get_serializer(containers, many=True)
+            return Response(serializer.data, status=status.HTTP_200_OK)
+        except Exception as e:
+            return Response({'detail': str(e)}, status=status.HTTP_503_SERVICE_UNAVAILABLE)
+
 
 class DomainViewSet(AppResourceViewSet):
     """A viewset for interacting with Domain objects."""

controller/scheduler/fleet.py

@@ -188,13 +188,20 @@ def _wait_for_container_state(self, name):
             raise RuntimeError('container timeout while retrieving state')
 
     def _wait_for_container_running(self, name):
+        # we bump to 20 minutes here to match the timeout on the router and in the app unit files
+        try:
+            self._wait_for_job_state(name, JobState.up)
+        except RuntimeError:
+            raise RuntimeError('container failed to start')
+
+    def _wait_for_job_state(self, name, state):
         # we bump to 20 minutes here to match the timeout on the router and in the app unit files
         for _ in range(1200):
-            if self.state(name) == JobState.up:
+            if self.state(name) == state:
                 return
             time.sleep(1)
         else:
-            raise RuntimeError('container failed to start')
+            raise RuntimeError('timeout waiting for job state: {}'.format(state))
 
     def _wait_for_destroy(self, name):
         for _ in range(30):
@@ -206,7 +213,8 @@ def _wait_for_destroy(self, name):
 
     def stop(self, name):
         """Stop a container"""
-        raise NotImplementedError
+        self._put_unit(name, {"desiredState": "loaded"})
+        self._wait_for_job_state(name, JobState.created)
 
     def destroy(self, name):
         """Destroy a container"""

controller/scheduler/mock.py

@@ -1,6 +1,6 @@
 import json
 from cStringIO import StringIO
-from .states import JobState, TransitionNotAllowed
+from .states import JobState, TransitionError
 
 
 # HACK: MockSchedulerClient is not persistent across requests
@@ -55,8 +55,14 @@ def start(self, name):
         """
         Start a container
         """
-        if self.state(name) not in [JobState.created, JobState.up, JobState.down]:
-            raise TransitionNotAllowed
+        if self.state(name) not in [JobState.created,
+                                    JobState.up,
+                                    JobState.down,
+                                    JobState.crashed,
+                                    JobState.error]:
+            raise TransitionError(self.state(name),
+                                  JobState.up,
+                                  'the container must be stopped or up to start')
         job = jobs.get(name, {})
         job.update({'state': JobState.up})
         jobs[name] = job
@@ -77,9 +83,11 @@ def stop(self, name):
         Stop a container
         """
         job = jobs.get(name, {})
-        if job.get('state') != JobState.up:
-            raise TransitionNotAllowed
-        job.update({'state': JobState.stopped})
+        if job.get('state') not in [JobState.up, JobState.crashed, JobState.error]:
+            raise TransitionError(job.get('state'),
+                                  JobState.up,
+                                  'the container must be up to stop')
+        job.update({'state': JobState.down})
         jobs[name] = job
         return
 

controller/scheduler/states.py

@@ -1,9 +1,14 @@
 import enum
 
 
-class TransitionNotAllowed(Exception):
+class TransitionError(Exception):
     """Raised when a transition from one state to another is illegal"""
 
+    def __init__(self, prev, next, msg):
+        self.prev = prev
+        self.next = next
+        self.msg = msg
+
 
 class JobState(enum.Enum):
     initialized = 1

tests/ps_test.go

@@ -16,13 +16,15 @@ var (
 	psListCmd      = "ps:list --app={{.AppName}}"
 	psScaleCmd     = "ps:scale web={{.ProcessNum}} --app={{.AppName}}"
 	psDownScaleCmd = "ps:scale web=0 --app={{.AppName}}"
+	psRestartCmd   = "ps:restart web --app={{.AppName}}"
 )
 
 func TestPs(t *testing.T) {
 	params := psSetup(t)
 	psScaleTest(t, params, psScaleCmd)
 	appsOpenTest(t, params)
 	psListTest(t, params, false)
+	psScaleTest(t, params, psRestartCmd)
 	psScaleTest(t, params, psDownScaleCmd)
 
 	// FIXME if we don't wait here, some of the routers may give us a 502 before