Fix #70 -- first pass at deis run <command> including test coverage

runs commands in ephemeral docker containers on runtime nodes.
no support for interactive shell yet (ie bash)

Author: mboersma

api/models.py

@@ -447,6 +447,13 @@ def logs(self):
         data = subprocess.check_output(['tail', '-n', str(settings.LOG_LINES), path])
         return data
 
+    def run(self, commands):
+        """Run a one-off command in an ephemeral container."""
+        runtime_nodes = self.node_set.filter(layer__id='runtime').order_by('?')
+        if not runtime_nodes:
+            raise EnvironmentError('No nodes available')
+        return runtime_nodes[0].run(commands)
+
     def destroy(self):
         """Create subtasks to terminate all nodes in parallel."""
         all_layers = self.layer_set.all()
@@ -625,6 +632,21 @@ def _prepare_terminate_args(self):
         args = (self.uuid, creds, params, self.provider_id)
         return args
 
+    def run(self, *args, **kwargs):
+        tasks = import_tasks(self.layer.flavor.provider.type)
+        command = ' '.join(*args)
+        # prepare app-specific docker arguments
+        formation_id = self.formation.id
+        release = self.formation.release_set.order_by('-created')[0]
+        version = release.version
+        docker_args = ' '.join(
+            ['-v',
+             '/opt/deis/runtime/slugs/{formation_id}-{version}/app:/app'.format(**locals()),
+             release.image])
+        args = list(self._prepare_converge_args()) + [docker_args] + [command]
+        task = tasks.run_node.subtask(args)
+        return task.apply_async().wait()
+
     def destroy(self, async=False):
         subtask = self.terminate()
         if async:

api/tests/formation.py

@@ -108,6 +108,19 @@ def test_formation_actions(self):
         response = self.client.post(url, json.dumps(body), content_type='application/json')
         self.assertEqual(response.status_code, 201)
         formation_id = response.data['id']  # noqa
+        # create & scale a basic formation
+        url = '/api/formations/{formation_id}/layers'.format(**locals())
+        body = {'id': 'proxy', 'flavor': 'autotest', 'run_list': 'recipe[deis::proxy]'}
+        response = self.client.post(url, json.dumps(body), content_type='application/json')
+        self.assertEqual(response.status_code, 201)
+        url = '/api/formations/{formation_id}/layers'.format(**locals())
+        body = {'id': 'runtime', 'flavor': 'autotest', 'run_list': 'recipe[deis::runtime]'}
+        response = self.client.post(url, json.dumps(body), content_type='application/json')
+        self.assertEqual(response.status_code, 201)
+        url = '/api/formations/{formation_id}/scale/layers'.format(**locals())
+        body = {'proxy': 2, 'runtime': 4}
+        response = self.client.post(url, json.dumps(body), content_type='application/json')
+        self.assertEqual(response.status_code, 200)
         # test calculate
         url = '/api/formations/{formation_id}/calculate'.format(**locals())
         response = self.client.post(url)
@@ -150,6 +163,13 @@ def test_formation_actions(self):
         response = self.client.post(url)
         self.assertEqual(response.status_code, 200)
         self.assertEqual(response.data, FAKE_LOG_DATA)
+        # test run
+        url = '/api/formations/{formation_id}/run'.format(**locals())
+        body = {'commands': 'ls -al'}
+        response = self.client.post(url, json.dumps(body), content_type='application/json')
+        self.assertEqual(response.status_code, 200)
+        self.assertIn('.gitignore', response.data[0])
+        self.assertEqual(response.data[1], 0)
 
 
 FAKE_LOG_DATA = """

api/urls.py

@@ -199,6 +199,12 @@
   See also
   :meth:`FormationViewSet.logs() <api.views.FormationViewSet.logs>`
 
+.. http:post:: /api/formations/(string:id)/run/
+
+  See also
+  :meth:`FormationViewSet.run() <api.views.FormationViewSet.run>`
+
+
 Auth
 ====
 
@@ -293,6 +299,8 @@
         views.FormationViewSet.as_view({'post': 'converge'})),
     url(r'^formations/(?P<id>[a-z0-9-]+)/logs/?',
         views.FormationViewSet.as_view({'post': 'logs'})),
+    url(r'^formations/(?P<id>[a-z0-9-]+)/run/?',
+        views.FormationViewSet.as_view({'post': 'run'})),
     # formation base endpoint
     url(r'^formations/(?P<id>[a-z0-9-]+)/?',
         views.FormationViewSet.as_view({'get': 'retrieve', 'delete': 'destroy'})),

api/views.py

@@ -234,6 +234,12 @@ def logs(self, request, **kwargs):
         return Response(logs, status=status.HTTP_200_OK,
                         content_type='text/plain')
 
+    def run(self, request, **kwargs):
+        formation = self.get_object()
+        output_and_rc = formation.run(request.DATA['commands'])
+        return Response(output_and_rc, status=status.HTTP_200_OK,
+                        content_type='text/plain')
+
     def destroy(self, request, **kwargs):
         formation = self.get_object()
         formation.destroy()

celerytasks/ec2.py

@@ -159,6 +159,14 @@ def converge_node(node_id, ssh_username, fqdn, ssh_private_key,
     return output, rc
 
 
+@task(name='ec2.run_node')
+def run_node(node_id, ssh_username, fqdn, ssh_private_key, docker_args, command):
+    ssh = util.connect_ssh(ssh_username, fqdn, 22, ssh_private_key)
+    command = "sudo docker run {docker_args} {command}".format(**locals())
+    output, rc = util.exec_ssh(ssh, command, pty=True)
+    return output, rc
+
+
 # utility functions
 
 def create_ec2_connection(region, access_key, secret_key):

celerytasks/mock.py

@@ -40,3 +40,23 @@ def converge_node(node_id, ssh_username, fqdn, ssh_private_key):
     output = ""
     rc = 0
     return output, rc
+
+
+@task(name='mock.run_node')
+def run_node(node_id, ssh_username, fqdn, ssh_private_key, docker_args, command):
+    output = """\
+total 80
+drwxr-xr-x  11 matt  staff   374 Aug 14 10:57 .
+drwxr-xr-x  34 matt  staff  1156 Aug 19 12:15 ..
+drwxr-xr-x  14 matt  staff   476 Aug 20 09:48 .git
+-rw-r--r--   1 matt  staff     5 Aug 14 10:57 .gitignore
+-rw-r--r--   1 matt  staff    11 Aug 14 10:57 .ruby-version
+-rw-r--r--   1 matt  staff    67 Aug 14 10:57 Gemfile
+-rw-r--r--   1 matt  staff   277 Aug 14 10:57 Gemfile.lock
+-rw-r--r--   1 matt  staff   553 Aug 14 10:57 LICENSE
+-rw-r--r--   1 matt  staff    37 Aug 14 10:57 Procfile
+-rw-r--r--   1 matt  staff  9165 Aug 14 10:57 README.md
+-rw-r--r--   1 matt  staff   127 Aug 14 10:57 web.rb
+"""
+    rc = 0
+    return output, rc

celerytasks/util.py

@@ -25,11 +25,12 @@ def connect_ssh(username, hostname, port, key):
     return ssh
 
 
-def exec_ssh(ssh, command):
+def exec_ssh(ssh, command, pty=False):
     tran = ssh.get_transport()
     chan = tran.open_session()
     # NOTE: pty breaks line ordering on commands like apt-get
-    #chan.get_pty(term='vt100', width=300, height=50)
+    if pty:
+        chan.get_pty(term='vt100', width=80, height=24)
     chan.exec_command(command)
     output = read_from_ssh(chan)
     exit_status = chan.recv_exit_status()
@@ -47,12 +48,12 @@ def read_from_ssh(chan):
                 if data:
                     got_data = True
                     output += data
-                    #print("stdout => ", data)
+                    # print("stdout => ", data)
             if chan.recv_stderr_ready():
                 data = r[0].recv_stderr(4096)
                 if data:
                     got_data = True
                     output += data
-                    #print("stderr => ", data)
+                    # print("stderr => ", data)
             if not got_data:
                 return output

client/deis.py

@@ -19,6 +19,7 @@
   converge      force-converge all nodes in the formation
   calculate     recalculate and update the formation databag
   logs          view aggregated log info for the formation
+  run           run a command on a remote container
   destroy       destroy a container formation
 
 Subcommands, use ``deis help [subcommand]`` to learn more::
@@ -644,6 +645,7 @@ def formations(self, args):
         formations:converge      force-converge all nodes in the formation
         formations:calculate     recalculate and update the formation databag
         formations:logs          view aggregated log info for the formation
+        formations:run           run a command on a remote container
         formations:destroy       destroy a container formation
 
         Use `deis help [command]` to learn more
@@ -835,6 +837,29 @@ def formations_converge(self, args):
         else:
             print('Error!', response.text)
 
+    def formations_run(self, args):
+        """
+        Run a command on a remote node.
+
+        Usage: deis formations:run <command>...
+        """
+        formation = args.get('--formation')
+        if not formation:
+            formation = self._session.formation
+        body = {'commands': sys.argv[2:]}
+        response = self._dispatch('post',
+                                  "/api/formations/{}/run".format(formation),
+                                  json.dumps(body))
+        if response.status_code == requests.codes.ok:  # @UndefinedVariable
+            output, rc = json.loads(response.content)
+            if rc == 0:
+                sys.stdout.write(output)
+                sys.stdout.flush()
+            else:
+                print('Error!\n{}'.format(output))
+        else:
+            print('Error!', response.text)
+
     def keys(self, args):
         """
         Valid commands for SSH keys:
@@ -1335,6 +1360,7 @@ def parse_args(cmd):
         'calculate': 'formations:calculate',
         'converge': 'formations:converge',
         'destroy': 'formations:destroy',
+        'run': 'formations:run',
         'scale': 'containers:scale',
         'ps': 'containers:list',
     }
@@ -1376,7 +1402,8 @@ def main():
                 return
         docopt(__doc__, argv=['--help'])
     # re-parse docopt with the relevant docstring
-    if cmd in dir(cli):
+    # unless cmd is formations_run, which needs to use sys.argv directly
+    if not cmd == 'formations_run' and cmd in dir(cli):
         docstring = trim(getattr(cli, cmd).__doc__)
         if 'Usage: ' in docstring:
             args.update(docopt(docstring))