Refactored API - added run_command as the simplest method of using the library. Output is returned as generators by default. Added tests for new functions. Resolves #14

Author: Dan

README.rst

@@ -42,10 +42,10 @@ Run `ls` on two remote hosts in parallel.
 >>> from pssh import ParallelSSHClient
 >>> hosts = ['myhost1', 'myhost2']
 >>> client = ParallelSSHClient(hosts)
->>> cmds = client.exec_command('ls -ltrh /tmp/aasdfasdf', sudo = True)
->>> print [client.get_stdout(cmd) for cmd in cmds]
+>>> output = client.run_command('ls -ltrh /tmp/aasdfasdf', sudo=True)
+>>> for host in output: print output
 [localhost]     drwxr-xr-x  6 xxx xxx 4.0K Jan  1 00:00 xxx
-[{'localhost': {'exit_code': 0}}]
+{'localhost': {'exit_code': 0, 'stdout': <generator>, 'stderr': <generator>, 'channel': channel}}
 
 **************************
 Frequently asked questions
@@ -84,6 +84,14 @@ Frequently asked questions
   >>> my_key = paramiko.RSAKey.from_private_key_file(my_rsa_key)
   >>> client = ParallelSSHClient(pkey=my_key)
 
+:Q:
+   Why should I use this module and not, for example, `fabric <https://github.com/fabric/fabric>`_?
+
+:A:
+   Fabric is a port of `capistrano <https://github.com/capistrano/capistrano>`_ from ruby to python. Its design goals are to provide a faithful port of capistrano with capistrano's `tasks` and `roles` to python with interactive command line being the intended usage - its use as a library is non-standard and in many cases just plain broken.
+   Furthermore, its parallel commands use a combination of both threads and processes with extremely high CPU usage while its running. Fabric currently stands at more than 130,000 lines of code, a large proportion of which is untested, particularly if used as a library as opposed to less than 700 currently in `ParallelSSH` with over 70% code test coverage.
+   ParallelSSH's design goals are to provide a *library* for running *asynchronous* SSH commands with **minimal** load induced on the system by doing so with the inteded usage being completely programmatic and non-interactive - Fabric provides none of these goals.
+
 ********
 SFTP/SCP
 ********

pssh.py

@@ -28,6 +28,7 @@
 See :mod:`pssh.ParallelSSHClient` and :mod:`pssh.SSHClient` for class documentation.
 """
 
+import warnings
 from socket import gaierror as sock_gaierror, error as sock_error
 import logging
 import paramiko
@@ -36,7 +37,7 @@
 from gevent import monkey
 monkey.patch_all()
 
-host_logger = logging.getLogger('host_logging')
+host_logger = logging.getLogger('host_logger')
 handler = logging.StreamHandler()
 host_log_format = logging.Formatter('%(message)s')
 handler.setFormatter(host_log_format)
@@ -213,8 +214,9 @@ def exec_command(self, command, sudo=False, user=None, **kwargs):
         channel.get_pty()
         _stdout, _stderr = channel.makefile('rb'), \
                            channel.makefile_stderr('rb')
-        stdout, stderr = self._read_output_buffer(_stdout), \
-                         self._read_output_buffer(_stderr)
+        stdout, stderr = self._read_output_buffer(_stdout,), \
+                         self._read_output_buffer(_stderr,
+                                                  prefix='\t[err]')
         if sudo and not user:
             command = 'sudo -S bash -c "%s"' % command.replace('"', '\\"')
         elif user:
@@ -229,12 +231,12 @@ def exec_command(self, command, sudo=False, user=None, **kwargs):
             gevent.sleep(.2)
         return channel, self.host, stdout, stderr
 
-    def _read_output_buffer(self, output_buffer):
-        """Read from output buffers,
-        allowing coroutines to execute in between reading"""
+    def _read_output_buffer(self, output_buffer, prefix=''):
+        """Read from output buffers and log to host_logger"""
         for line in output_buffer:
-            gevent.sleep()
-            yield line.strip()
+            output = line.strip()
+            host_logger.info("[%s]%s\t%s", self.host, prefix, output,)
+            yield output
 
     def _make_sftp(self):
         """Make SFTP client from open transport"""
@@ -404,9 +406,38 @@ def __init__(self, hosts,
         # To hold host clients
         self.host_clients = dict((host, None) for host in hosts)
 
+    def run_command(self, *args, **kwargs):
+        """Run command on all hosts in parallel, honoring self.pool_size
+
+        :param args: Position arguments for command
+        :type args: tuple
+        :param kwargs: Keyword arguments for command
+        :type kwargs: dict
+
+        :rtype: List of :mod:`gevent.Greenlet`
+
+        **Example**:
+        
+        >>> output = client.exec_command('ls -ltrh')
+        
+        Wait for completion, no stdout:
+        
+        >>> client.pool.join()
+        
+        Alternatively/in addition print stdout for each command:
+        
+        >>> for host in output:
+        >>>     print output[host]['stdout']
+        """
+        for host in self.hosts:
+            self.pool.spawn(self._exec_command, host, *args, **kwargs)
+        return self.get_output()
+    
     def exec_command(self, *args, **kwargs):
         """Run command on all hosts in parallel, honoring self.pool_size
 
+        **Superseeded by :mod:`ParallelSSH.run_command`**
+
         :param args: Position arguments for command
         :type args: tuple
         :param kwargs: Keyword arguments for command
@@ -446,35 +477,80 @@ def _exec_command(self, host, *args, **kwargs):
                                                 timeout=self.timeout)
         return self.host_clients[host].exec_command(*args, **kwargs)
 
+    def get_output(self, commands=None):
+        """Get output from running commands.
+
+        Stdout and stderr are also logged via the logger named ``host_logger``
+        which is enabled by default.
+        ``host_logger`` output can be disabled by removing its handler.
+        >>> logger = logging.getLogger('host_logger')
+        >>> for handler in logger.handlers: logger.removeHandler(handler)
+
+        **Example usage**:
+
+        >>> output = client.get_output()
+        >>> for host in output: print output[host]['stdout']
+        <stdout>
+        >>> # Get exit code after command has finished
+        >>> self.get_exit_code(output[host])
+        0
+
+        :param commands: (Optional) Override commands to get output from.
+        Uses running commands in pool if not given
+        :type commands: :mod:`gevent.Greenlet`
+        :rtype: Dictionary with host as key as in:
+        ``{'myhost1': {'exit_code': exit code if ready else None,
+                       'channel' : SSH channel of command,
+                       'stdout' : <iterable>,
+                       'stderr' : <iterable>,}}``"""
+        if not commands:
+            commands = list(self.pool.greenlets)
+        return {host: {'exit_code': self._get_exit_code(channel),
+                       'channel' : channel,
+                       'stdout' : stdout,
+                       'stderr' : stderr, }
+                       for cmd in commands
+                       for (channel, host, stdout, stderr) in [cmd.get()]}
+
+    def get_exit_code(self, host_output):
+        """Get exit code from host output if available
+        :param host_output: Per host output as returned by `self.get_output`
+        :rtype: int or None if exit code not ready"""
+        if not 'channel' in host_output:
+            logger.error("%s does not look like host output..", host_output,)
+            return
+        channel = host_output['channel']
+        return self._get_exit_code(channel)
+
+    def _get_exit_code(self, channel):
+        """Get exit code from channel if ready"""
+        if not channel.exit_status_ready():
+            return
+        channel.close()
+        return channel.recv_exit_status()
+
     def get_stdout(self, greenlet, return_buffers=False):
         """Get/print stdout from greenlet and return exit code for host
         
-        :mod:`pssh.get_stdout` will close the open SSH channel but this does
-        **not** close the established connection to the remote host, only the
-        authenticated SSH channel within it. This is standard practise
-        in SSH when a command has finished executing. A new command
-        will open a new channel which is very fast on already established
-        connections.
-
-        By default, stdout and stderr will be logged via the logger named \
-        ``host_logger`` unless ``return_buffers`` is set to ``True`` in which case \
-        both buffers are instead returned along with the exit status.
-
+        **Deprecated** - use self.get_output() instead.
+        
         :param greenlet: Greenlet object containing an \
         SSH channel reference, hostname, stdout and stderr buffers
         :type greenlet: :mod:`gevent.Greenlet`
-
+        
         :param return_buffers: Flag to turn on returning stdout and stderr \
         buffers along with exit code. Defaults to off.
         :type return_buffers: bool
-
+        
         :rtype: Dictionary containing ``{host: {'exit_code': exit code}}`` entry \
         for example ``{'myhost1': {'exit_code': 0}}``
         :rtype: With ``return_buffers=True``: ``{'myhost1': {'exit_code': 0,
                                                              'channel' : None or SSH channel of command if command is still executing,
                                                              'stdout' : <iterable>,
                                                              'stderr' : <iterable>,}}``
         """
+        warnings.warn("This method is being deprecated and will be removed in \
+future releases - use self.get_output instead", DeprecationWarning)
         gevent.sleep(.2)
         channel, host, stdout, stderr = greenlet.get()
         if channel.exit_status_ready():
@@ -496,16 +572,6 @@ def get_stdout(self, greenlet, return_buffers=False):
                        'stdout' : stdout,
                        'stderr' : stderr, }}
 
-    # WIP
-    # def wait_for_exit_status(self, channel):
-    #     """Block and wait for exit status on channel.
-    #     WARNING - this will block forever if the command executed never exits
-    #     :rtype: int - Exit code of command executed"""
-    #     while not channel.exit_status_ready():
-    #         gevent.sleep()
-    #     channel.close()
-    #     return channel.recv_exit_status()
-
     def copy_file(self, local_file, remote_file):
         """Copy local file to remote file in parallel
         

tests/test_pssh_client.py

@@ -96,6 +96,48 @@ def test_pssh_client_exec_command_get_buffers(self):
         del client
         server.join()
 
+    def test_pssh_client_run_command_get_output(self):
+        server = start_server({ self.fake_cmd : self.fake_resp }, self.listen_socket)
+        client = ParallelSSHClient(['127.0.0.1'], port=self.listen_port,
+                                   pkey=self.user_key)
+        output = client.run_command(self.fake_cmd)
+        # import ipdb; ipdb.set_trace()
+        expected_exit_code = 0
+        expected_stdout = [self.fake_resp]
+        expected_stderr = []
+        exit_code = output['127.0.0.1']['exit_code']
+        stdout = list(output['127.0.0.1']['stdout'])
+        stderr = list(output['127.0.0.1']['stderr'])
+        self.assertEqual(expected_exit_code, exit_code,
+                         msg = "Got unexpected exit code - %s, expected %s" %
+                         (exit_code,
+                          expected_exit_code,))
+        self.assertEqual(expected_stdout, stdout,
+                         msg = "Got unexpected stdout - %s, expected %s" % 
+                         (stdout,
+                          expected_stdout,))
+        self.assertEqual(expected_stderr, stderr,
+                         msg = "Got unexpected stderr - %s, expected %s" % 
+                         (stderr,
+                          expected_stderr,))
+        del client
+        server.join()
+
+    def test_pssh_client_run_long_command(self):
+        expected_lines = 5
+        server = start_server({ self.long_running_cmd :
+                                self.long_running_response(expected_lines) },
+                              self.listen_socket)
+        client = ParallelSSHClient(['127.0.0.1'], port=self.listen_port,
+                                   pkey=self.user_key)
+        output = client.run_command(self.long_running_cmd)
+        self.assertTrue('127.0.0.1' in output, msg="Got no output for command")
+        stdout = list(output['127.0.0.1']['stdout'])
+        self.assertTrue(len(stdout) == expected_lines, msg="Expected %s lines of response, got %s" %
+                        (expected_lines, len(stdout)))
+        del client
+        server.kill()
+
     def test_pssh_client_auth_failure(self):
         server = start_server({ self.fake_cmd : self.fake_resp },
                               self.listen_socket, fail_auth=True)