Clush: added -P/--progress option

Added --progress option to clush to force display of the live progress
indicator and display global write bandwidth when writing standard input
(make use of the ev_written event handler).

Added documentation for clush progress indicator.

Change-Id: Iaa2b6613939792de9ada3d333c1c5cc5cc3144a9

Author: thiell

ChangeLog

@@ -1,3 +1,8 @@
+2015-11-01 S. Thiell <[email protected]>
+
+	* Clush.py: added -P/--progress to force display of the live progress
+	indicator and display global write bandwidth when writing standard input.
+
 2015-10-25 S. Thiell <[email protected]>
 
 	* Clush.py: added --option/-O clush.conf settings override (pull

doc/man/man1/clush.1

@@ -1,6 +1,6 @@
 .\" Man page generated from reStructuredText.
 .
-.TH CLUSH 1 "2015-10-30" "1.7" "ClusterShell User Manual"
+.TH CLUSH 1 "2015-11-01" "1.7" "ClusterShell User Manual"
 .SH NAME
 clush \- execute shell commands on a cluster
 .
@@ -213,6 +213,9 @@ disable header block and order output by nodes; additionally, when used in conju
 .B \-N
 disable labeling of command line
 .TP
+.B \-P\fP,\fB  \-\-progress
+show progress during command execution; if writing is performed to standard input, the live progress indicator will display the global bandwidth of data written to the target nodes
+.TP
 .B \-b\fP,\fB  \-\-dshbak
 display gathered results in a dshbak\-like way
 .TP

doc/sphinx/tools/clush.rst

@@ -228,14 +228,15 @@ specified remote hosts in parallel (given the current *fanout* value and the
 number of commands to execute (see *fanout* library settings in
 :ref:`class-Task-configure`).
 
+.. _clush-gather:
+
 Output gathering options
 """"""""""""""""""""""""
 
 If option ``-b`` or ``--dshbak`` is specified, *clush* waits for command
-completion while displaying a progress indicator (unless ``-q, --quiet``
-switch is provided) and then displays gathered output results. If standard
-output is redirected to a file, *clush* detects it and disable any progress
-indicator.
+completion while displaying a :ref:`progress indicator <clush-progress>` and
+then displays gathered output results. If standard output is redirected to a
+file, *clush* detects it and disable any progress indicator.
 
 The following is a simple example of *clush* command used to execute ``uname
 -r`` on *node40*, *node41* and *node42*, wait for their completion and finally
@@ -312,6 +313,32 @@ Another stdin-bound *clush* usage example::
     ---------------
     /etc/yum.repos.d/cobbler-config.repo
 
+.. _clush-progress:
+
+Progress indicator
+""""""""""""""""""
+
+In :ref:`output gathering mode <clush-gather>`, *clush* will display a live
+progress indicator as a simple but convenient way to follow the completion of
+parallel commands. It can be disabled just by using the ``-q`` or ``--quiet``
+options. The progress indicator will appear after 1 to 2 seconds and should
+look like this::
+
+    clush: <command_completed>/<command_total>
+
+If writing is performed to *clush* standard input, like in ``command |
+clush``, the live progress indicator will display the global bandwidth of data
+written to the target nodes.
+
+Finally, the special option ``--progress`` can be used to force the display of
+the live progress indicator. Using this option may interfere with some command
+outputs, but it can be useful when using stdin while remote commands are
+silent. As an example, the following command will copy a local file to
+node[1-3] and display the global write bandwidth to the target nodes::
+
+    $ dd if=/path/to/local/file | clush -w node[1-3] --progress 'dd of=/path/to/remote/file'
+    clush: 0/3 write: 212.27 MiB/s
+
 .. _clush-interactive:
 
 Interactive mode

doc/txt/clush.txt

@@ -7,7 +7,7 @@ execute shell commands on a cluster
 -----------------------------------
 
 :Author: Stephane Thiell <[email protected]>
-:Date:   2015-10-30
+:Date:   2015-11-01
 :Copyright: CeCILL-C V1
 :Version: 1.7
 :Manual section: 1
@@ -152,6 +152,7 @@ Output behaviour:
   -G, --groupbase     do not display group source prefix
   -L                  disable header block and order output by nodes; additionally, when used in conjunction with -b/-B, it will enable "life gathering" of results by line mode, such as the next line is displayed as soon as possible (eg. when all nodes have sent the line)
   -N                  disable labeling of command line
+  -P, --progress      show progress during command execution; if writing is performed to standard input, the live progress indicator will display the global bandwidth of data written to the target nodes
   -b, --dshbak        display gathered results in a dshbak-like way
   -B                  like -b but including standard error
   -r, --regroup       fold nodeset using node groups

lib/ClusterShell/CLI/Clush.py

@@ -51,6 +51,7 @@
 import resource
 import sys
 import signal
+import time
 import threading
 
 from ClusterShell.Defaults import DEFAULTS, _load_workerclass
@@ -59,7 +60,7 @@
 from ClusterShell.CLI.Display import VERB_QUIET, VERB_STD, VERB_VERB, VERB_DEBUG
 from ClusterShell.CLI.OptionParser import OptionParser
 from ClusterShell.CLI.Error import GENERIC_ERRORS, handle_generic_error
-from ClusterShell.CLI.Utils import NodeSet, bufnodeset_cmp
+from ClusterShell.CLI.Utils import NodeSet, bufnodeset_cmp, human_bi_bytes_unit
 
 from ClusterShell.Event import EventHandler
 from ClusterShell.MsgTree import MsgTree
@@ -92,10 +93,11 @@ def __init__(self):
         EventHandler.__init__(self)
         self._runtimer = None
 
-    def runtimer_init(self, task, ntotal):
+    def runtimer_init(self, task, ntotal=0):
         """Init timer for live command-completed progressmeter."""
-        self._runtimer = task.timer(2.0, RunTimer(task, ntotal),
-                                    interval=1./3., autoclose=True)
+        thandler = RunTimer(task, ntotal)
+        self._runtimer = task.timer(1.33, thandler, interval=1./3.,
+                                    autoclose=True)
 
     def _runtimer_clean(self):
         """Hide runtimer counter"""
@@ -124,6 +126,16 @@ def update_prompt(self, worker):
         if worker.task.default("USER_handle_SIGUSR1"):
             os.kill(os.getpid(), signal.SIGUSR1)
 
+    def ev_start(self, worker):
+        """Worker is starting."""
+        if self._runtimer:
+            self._runtimer.eh.start_time = time.time()
+
+    def ev_written(self, worker, node, sname, size):
+        """Bytes written on worker"""
+        if self._runtimer:
+            self._runtimer.eh.bytes_written += size
+
 class DirectOutputHandler(OutputHandler):
     """Direct output event handler class."""
 
@@ -156,7 +168,29 @@ def ev_timeout(self, worker):
     def ev_close(self, worker):
         self.update_prompt(worker)
 
-class CopyOutputHandler(DirectOutputHandler):
+class DirectProgressOutputHandler(DirectOutputHandler):
+    """Direct output event handler class with progress support."""
+
+    # NOTE: This class is very similar to DirectOutputHandler, thus it could
+    #       first look overkill, but merging both is slightly impacting ev_read
+    #       performance of current DirectOutputHandler.
+
+    def ev_read(self, worker):
+        self._runtimer_clean()
+        # it is ~10% faster to avoid calling super here
+        node = worker.current_node or worker.key
+        self._display.print_line(node, worker.current_msg)
+
+    def ev_error(self, worker):
+        self._runtimer_clean()
+        node = worker.current_node or worker.key
+        self._display.print_line_error(node, worker.current_errmsg)
+
+    def ev_close(self, worker):
+        self._runtimer_clean()
+        DirectOutputHandler.ev_close(self, worker)
+
+class CopyOutputHandler(DirectProgressOutputHandler):
     """Copy output event handler."""
     def __init__(self, display, reverse=False):
         DirectOutputHandler.__init__(self, display)
@@ -308,6 +342,9 @@ def __init__(self, task, total):
         self.tslen = len(str(self.total))
         self.wholelen = 0
         self.started = False
+        # updated by worker handler for progress
+        self.start_time = 0
+        self.bytes_written = 0
 
     def ev_timer(self, timer):
         self.update()
@@ -318,8 +355,15 @@ def set_dirty(self):
     def erase_line(self):
         if self.wholelen:
             sys.stderr.write(' ' * self.wholelen + '\r')
+            self.wholelen = 0
 
     def update(self):
+        """Update runtime progress info"""
+        wrbwinfo = ''
+        if self.bytes_written > 0:
+            bandwidth = self.bytes_written/(time.time() - self.start_time)
+            wrbwinfo = " write: %s/s" % human_bi_bytes_unit(bandwidth)
+
         gws = self.task.gateways.keys()
         if gws:
             # tree mode
@@ -331,11 +375,12 @@ def update(self):
         else:
             cnt = len(self.task._engine.clients())
             gwinfo = ''
-        if cnt != self.cnt_last:
+        if self.bytes_written > 0 or cnt != self.cnt_last:
             self.cnt_last = cnt
             # display completed/total clients
-            towrite = 'clush: %*d/%*d%s\r' % (self.tslen, self.total - cnt,
-                                              self.tslen, self.total, gwinfo)
+            towrite = 'clush: %*d/%*d%s%s\r' % (self.tslen, self.total - cnt,
+                                                self.tslen, self.total, gwinfo,
+                                                wrbwinfo)
             self.wholelen = len(towrite)
             sys.stderr.write(towrite)
             self.started = True
@@ -344,6 +389,7 @@ def finalize(self, force_cr):
         """finalize display of runtimer"""
         if not self.started:
             return
+        self.erase_line()
         # display completed/total clients
         fmt = 'clush: %*d/%*d'
         if force_cr:
@@ -599,9 +645,14 @@ def run_command(task, cmd, ns, timeout, display, remote):
         else:
             handler = GatherOutputHandler(display)
 
-        if display.verbosity == VERB_STD or display.verbosity == VERB_VERB:
+        if display.verbosity in (VERB_STD, VERB_VERB) or \
+            (display.progress and display.verbosity > VERB_QUIET):
             handler.runtimer_init(task, len(ns))
+    elif display.progress and display.verbosity > VERB_QUIET:
+        handler = DirectProgressOutputHandler(display)
+        handler.runtimer_init(task, len(ns))
     else:
+        # this is the simpler but faster output handler
         handler = DirectOutputHandler(display)
 
     worker = task.shell(cmd, nodes=ns, handler=handler, timeout=timeout,
@@ -614,9 +665,7 @@ def run_command(task, cmd, ns, timeout, display, remote):
     task.resume()
 
 def run_copy(task, sources, dest, ns, timeout, preserve_flag, display):
-    """
-    run copy command
-    """
+    """run copy command"""
     task.set_default("USER_running", True)
     task.set_default("USER_copies", len(sources))
 
@@ -626,7 +675,7 @@ def run_copy(task, sources, dest, ns, timeout, preserve_flag, display):
         print Display.COLOR_RESULT_FMT % '-' * 15
 
     copyhandler = CopyOutputHandler(display)
-    if display.verbosity == VERB_STD or display.verbosity == VERB_VERB:
+    if display.verbosity in (VERB_STD, VERB_VERB):
         copyhandler.runtimer_init(task, len(ns) * len(sources))
 
     # Sources check
@@ -640,9 +689,7 @@ def run_copy(task, sources, dest, ns, timeout, preserve_flag, display):
     task.resume()
 
 def run_rcopy(task, sources, dest, ns, timeout, preserve_flag, display):
-    """
-    run reverse copy command
-    """
+    """run reverse copy command"""
     task.set_default("USER_running", True)
     task.set_default("USER_copies", len(sources))
 

lib/ClusterShell/CLI/Display.py

@@ -1,7 +1,7 @@
 #!/usr/bin/env python
 #
-# Copyright CEA/DAM/DIF (2010, 2011, 2012)
-#  Contributor: Stephane THIELL <[email protected]>
+# Copyright CEA/DAM/DIF (2010-2015)
+#  Contributor: Stephane THIELL <[email protected]>
 #
 # This file is part of the ClusterShell library.
 #
@@ -82,6 +82,7 @@ def __init__(self, options, config=None, color=None):
         self._diffref = None
         # diff implies at least -b
         self.gather = options.gatherall or options.gather or options.diff
+        self.progress = getattr(options, 'progress', False) # only in clush
         # check parameter combinaison
         if options.diff and options.line_mode:
             raise ValueError("diff not supported in line_mode")

lib/ClusterShell/CLI/OptionParser.py

@@ -158,6 +158,8 @@ def install_display_options(self,
             optgrp.add_option("-b", "-c", "--dshbak", action="store_true",
                 dest="gather", help="gather nodes with same output")
         else:
+            optgrp.add_option("-P", "--progress", action="store_true",
+                dest="progress", help="show progress during command execution")
             optgrp.add_option("-b", "--dshbak", action="store_true",
                 dest="gather", help="gather nodes with same output")
         optgrp.add_option("-B", action="store_true", dest="gatherall",

lib/ClusterShell/CLI/Utils.py

@@ -1,7 +1,7 @@
 #!/usr/bin/env python
 #
-# Copyright CEA/DAM/DIF (2010)
-#  Contributor: Stephane THIELL <[email protected]>
+# Copyright CEA/DAM/DIF (2010-2015)
+#  Contributor: Stephane THIELL <[email protected]>
 #
 # This file is part of the ClusterShell library.
 #
@@ -46,6 +46,24 @@
         "ERROR: ClusterShell node groups configuration error:\n\t%s" % exc
     sys.exit(1)
 
+(KIBI, MEBI, GIBI, TEBI) = (1024.0, 1024.0 ** 2, 1024.0 ** 3, 1024.0 ** 4)
+
+def human_bi_bytes_unit(value):
+    """
+    Format numerical `value` to display it using human readable unit with
+    binary prefix like (KiB, MiB, GiB, ...).
+    """
+    if value >= TEBI:
+        fmt = "%.2f TiB" % (value / TEBI)
+    elif value >= GIBI:
+        fmt = "%.2f GiB" % (value / GIBI)
+    elif value >= MEBI:
+        fmt = "%.2f MiB" % (value / MEBI)
+    elif value >= KIBI:
+        fmt = "%.2f KiB" % (value / KIBI)
+    else:
+        fmt = "%d B" % value
+    return fmt
 
 def nodeset_cmp(ns1, ns2):
     """Compare 2 nodesets by their length (we want larger nodeset
@@ -65,4 +83,3 @@ def bufnodeset_cmp(bn1, bn2):
     node."""
     # Extract nodesets and call nodeset_cmp
     return nodeset_cmp(bn1[1], bn2[1])
-

tests/CLIClushTest.py

@@ -427,6 +427,38 @@ def test_030_config_options(self):
                        "-O", "color=never", "-w", HOSTNAME, "echo", "ok"],
                       None, "%s: ok\n" % HOSTNAME)
 
+    def test_031_progress(self):
+        """test clush -P/--progress"""
+        self._clush_t(["-w", HOSTNAME, "--progress", "echo", "ok"], None,
+                      "%s: ok\n" % HOSTNAME)
+        self._clush_t(["-w", HOSTNAME, "--progress", "sleep", "2"], None, '', 0,
+                      re.compile(r'clush: 0/1\r.*'))
+        self._clush_t(["-w", HOSTNAME, "--progress", "sleep", "2"], 'AAAAAAAA',
+                      '', 0, re.compile(r'clush: 0/1 write: \d B/s\r.*'))
+        self._clush_t(["-w", "%s,localhost" % HOSTNAME, "--progress",
+                       "sleep", "2"], 'AAAAAAAAAAAAAA', '', 0,
+                       re.compile(r'clush: 0/2 write: \d+ B/s\r.*'))
+        self._clush_t(["-w", HOSTNAME, "-b", "--progress", "sleep", "2"],
+                      None, '', 0, re.compile(r'clush: 0/1\r.*'))
+        self._clush_t(["-w", HOSTNAME, "-b", "--progress", "sleep", "2"],
+                      'AAAAAAAAAAAAAAAA', '', 0,
+                      re.compile(r'clush: 0/1 write: \d+ B/s\r.*'))
+        # -q and --progress: explicit -q wins
+        self._clush_t(["-w", HOSTNAME, "--progress", "-q", "sleep", "2"], None,
+                      '', 0)
+        self._clush_t(["-w", HOSTNAME, "-b", "--progress", "-q", "sleep", "2"],
+                      None, '', 0, '')
+        self._clush_t(["-w", HOSTNAME, "-b", "--progress", "-q", "sleep", "2"],
+                      'AAAAAAAAAAAAAAAA', '', 0, '')
+        # cover stderr output and --progress
+        self._clush_t(["-w", HOSTNAME, "--progress",
+                       "echo foo >&2; echo bar; sleep 2"], None,
+                      "%s: bar\n" % HOSTNAME, 0,
+                      re.compile(r'%s: foo\nclush: 0/1\r.*'
+                                 % HOSTNAME))
+
+
+
 class CLIClushTest_B_StdinFailure(unittest.TestCase):
     """Unit test class for testing CLI/Clush.py and stdin failure"""