Worker/Tree.py: implement reverse file copy

Implement tree mode reverse copy using tar commands. Principle is
similar to the one used for regular copy, but tar data are transferred
back through stdout (instead of sent through stdin). Data is also
encoded using base64. To ease tar extraction on head node, each hostname
is added as suffix to the base file (or directory) by tar itself, hence
the complicated command.

Closes #290 (fix clush --rcopy).

Change-Id: Ia7a7ab4ab513520b46fbea96836243497fc2c94a

Author: thiell

ChangeLog

@@ -1,5 +1,8 @@
 2016-02-27 S. Thiell <[email protected]>
 
+	* Worker/Tree.py: implement tree mode reverse copy using tar commands to
+	fix clush --rcopy (ticket #290).
+
 	* Communication.py: remove 76-char base64 encoding fixed length
 	restriction for tree XML payload communication. The default max length is
 	now 64K, which gives good results. The environment variable

lib/ClusterShell/CLI/Clush.py

@@ -640,11 +640,6 @@ def run_command(task, cmd, ns, timeout, display, remote):
     """
     task.set_default("USER_running", True)
 
-    if display.verbosity >= VERB_VERB and task.topology:
-        print Display.COLOR_RESULT_FMT % '-' * 15
-        print Display.COLOR_RESULT_FMT % task.topology,
-        print Display.COLOR_RESULT_FMT % '-' * 15
-
     if (display.gather or display.line_mode) and ns is not None:
         if display.gather and display.line_mode:
             handler = LiveGatherOutputHandler(display, ns)
@@ -675,20 +670,15 @@ def run_copy(task, sources, dest, ns, timeout, preserve_flag, display):
     task.set_default("USER_running", True)
     task.set_default("USER_copies", len(sources))
 
-    if display.verbosity >= VERB_VERB and task.topology:
-        print Display.COLOR_RESULT_FMT % '-' * 15
-        print Display.COLOR_RESULT_FMT % task.topology,
-        print Display.COLOR_RESULT_FMT % '-' * 15
-
     copyhandler = CopyOutputHandler(display)
     if display.verbosity in (VERB_STD, VERB_VERB):
         copyhandler.runtimer_init(task, len(ns) * len(sources))
 
     # Sources check
     for source in sources:
         if not exists(source):
-            display.vprint_err(VERB_QUIET, "ERROR: file \"%s\" not found" % \
-                                           source)
+            display.vprint_err(VERB_QUIET,
+                               'ERROR: file "%s" not found' % source)
             clush_exit(1, task)
         task.copy(source, dest, ns, handler=copyhandler, timeout=timeout,
                   preserve=preserve_flag)
@@ -701,20 +691,20 @@ def run_rcopy(task, sources, dest, ns, timeout, preserve_flag, display):
 
     # Sanity checks
     if not exists(dest):
-        display.vprint_err(VERB_QUIET, "ERROR: directory \"%s\" not found" % \
-                                       dest)
+        display.vprint_err(VERB_QUIET,
+                           'ERROR: directory "%s" not found' % dest)
         clush_exit(1, task)
     if not isdir(dest):
-        display.vprint_err(VERB_QUIET, \
-            "ERROR: destination \"%s\" is not a directory" % dest)
+        display.vprint_err(VERB_QUIET,
+                           'ERROR: destination "%s" is not a directory' % dest)
         clush_exit(1, task)
 
     copyhandler = CopyOutputHandler(display, True)
     if display.verbosity == VERB_STD or display.verbosity == VERB_VERB:
         copyhandler.runtimer_init(task, len(ns) * len(sources))
     for source in sources:
         task.rcopy(source, dest, ns, handler=copyhandler, timeout=timeout,
-                   preserve=preserve_flag)
+                   stderr=True, preserve=preserve_flag)
     task.resume()
 
 def set_fdlimit(fd_max, display):
@@ -977,9 +967,12 @@ def main():
 
     if options.grooming_delay:
         if config.verbosity >= VERB_VERB:
-            print Display.COLOR_RESULT_FMT % ("Grooming delay: %f" % \
-                                              options.grooming_delay)
+            print Display.COLOR_RESULT_FMT % ("Grooming delay: %f"
+                                              % options.grooming_delay)
         task.set_info("grooming_delay", options.grooming_delay)
+    elif options.rcopy:
+        # By default, --rcopy should inhibit grooming
+        task.set_info("grooming_delay", 0)
 
     if config.ssh_user:
         task.set_info("ssh_user", config.ssh_user)
@@ -1043,6 +1036,10 @@ def main():
                                                 config.command_timeout,
                                                 op))
     if not task.default("USER_interactive"):
+        if display.verbosity >= VERB_VERB and task.topology:
+            print Display.COLOR_RESULT_FMT % '-' * 15
+            print Display.COLOR_RESULT_FMT % task.topology,
+            print Display.COLOR_RESULT_FMT % '-' * 15
         if options.copy:
             run_copy(task, args, options.dest_path, nodeset_base, timeout,
                      options.preserve_flag, display)

lib/ClusterShell/Propagation.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env python
 #
-# Copyright CEA/DAM/DIF (2010-2015)
+# Copyright CEA/DAM/DIF (2010-2016)
 #  Contributor: Henri DOREAU <[email protected]>
 #  Contributor: Stephane THIELL <[email protected]>
 #
@@ -263,7 +263,7 @@ def start(self):
 
     def recv(self, msg):
         """process incoming messages"""
-        self.logger.debug("[DBG] rcvd from: %s", msg)
+        self.logger.debug("recv: %s", msg)
         if msg.type == EndMessage.ident:
             #??#self.ptree.notify_close()
             self.logger.debug("got EndMessage; closing")
@@ -344,7 +344,6 @@ def recv_cfg(self, msg):
 
     def recv_ctl(self, msg):
         """handle incoming messages for state 'control'"""
-        self.logger.debug("recv_ctl")
         if msg.type == 'ACK':
             self.logger.debug("got ack (%s)", msg.type)
             # check if ack matches write history msgid to generate ev_written
@@ -361,7 +360,8 @@ def recv_ctl(self, msg):
                 if metaworker.eh:
                     nodeset = NodeSet(msg.nodes)
                     decoded = msg.data_decode() + '\n'
-                    self.logger.debug("StdOutMessage: \"%s\"", decoded)
+                    # this is way too verbose for --rcopy
+                    #self.logger.debug("StdOutMessage: \"%s\"", decoded)
                     for line in decoded.splitlines():
                         for node in nodeset:
                             metaworker._on_remote_node_msgline(node,

lib/ClusterShell/Worker/Tree.py

@@ -34,6 +34,7 @@
 ClusterShell v2 tree propagation worker
 """
 
+import base64
 import logging
 import os
 from os.path import basename, dirname, isfile, normpath
@@ -67,8 +68,6 @@ def ev_read(self, worker):
         """
         Called to indicate that a worker has data to read.
         """
-        self.logger.debug("MetaWorkerEventHandler: ev_read (%s)",
-                          worker.current_sname)
         self.metaworker._on_node_msgline(worker.current_node,
                                          worker.current_msg,
                                          'stdout')
@@ -127,7 +126,12 @@ class WorkerTree(DistantWorker):
     ClusterShell tree worker Class.
 
     """
-    UNTAR_CMD_FMT = 'tar -xf - -C "%s"'
+    # copy and rcopy tar command formats
+    # the choice of single or double quotes is essential
+    UNTAR_CMD_FMT = "tar -xf - -C '%s'"
+    TAR_CMD_FMT = "tar -cf - -C '%s' " \
+                  "--transform \"s,^\\([^/]*\\)[/]*,\\1.$(hostname -s)/,\" " \
+                  "'%s' | base64 -w 65536"
 
     def __init__(self, nodes, handler, timeout, **kwargs):
         """
@@ -142,6 +146,7 @@ def __init__(self, nodes, handler, timeout, **kwargs):
         """
         DistantWorker.__init__(self, handler)
 
+        self.logger = logging.getLogger(__name__)
         self.workers = []
         self.nodes = NodeSet(nodes)
         self.timeout = timeout
@@ -150,18 +155,27 @@ def __init__(self, nodes, handler, timeout, **kwargs):
         self.dest = kwargs.get('dest')
         autoclose = kwargs.get('autoclose', False)
         self.stderr = kwargs.get('stderr', False)
+        self.logger.debug("stderr=%s", self.stderr)
         self.remote = kwargs.get('remote', True)
+        self.preserve = kwargs.get('preserve', None)
+        self.reverse = kwargs.get('reverse', False)
+        self._rcopy_bufs = {}
+        self._rcopy_tars = {}
         self._close_count = 0
         self._start_count = 0
         self._child_count = 0
         self._target_count = 0
         self._has_timeout = False
-        self.logger = logging.getLogger(__name__)
 
         if self.command is None and self.source is None:
             raise ValueError("missing command or source parameter in "
                              "WorkerTree constructor")
 
+        # rcopy is enforcing separated stderr to handle tar error messages
+        # because stdout is used for data transfer
+        if self.source and self.reverse:
+            self.stderr = True
+
         # build gateway invocation command
         invoke_gw_args = []
         for envname in ('PYTHONPATH',
@@ -183,6 +197,7 @@ def __init__(self, nodes, handler, timeout, **kwargs):
             self.router = None
 
         self.upchannel = None
+
         self.metahandler = MetaWorkerEventHandler(self)
 
         # gateway -> active targets selection
@@ -211,19 +226,29 @@ def _launch(self, nodes):
         # Prepare copy params if source is defined
         destdir = None
         if self.source:
-            self.logger.debug("copy self.dest=%s", self.dest)
-            # Special processing to determine best arcname and destdir for tar.
-            # The only case that we don't support is when source is a file and
-            # dest is a dir without a finishing / (in that case we cannot
-            # determine remotely whether it is a file or a directory).
-            if isfile(self.source):
-                # dest is not normalized here
-                arcname = basename(self.dest) or basename(normpath(self.source))
-                destdir = dirname(self.dest)
+            if self.reverse:
+                self.logger.debug("rcopy source=%s, dest=%s", self.source,
+                                  self.dest)
+                # dest is a directory
+                destdir = self.dest
             else:
-                arcname = basename(normpath(self.source))
-                destdir = os.path.normpath(self.dest)
-            self.logger.debug("copy arcname=%s destdir=%s", arcname, destdir)
+                self.logger.debug("copy source=%s, dest=%s", self.source,
+                                  self.dest)
+                # Special processing to determine best arcname and destdir for
+                # tar. The only case that we don't support is when source is a
+                # file and dest is a dir without a finishing / (in that case we
+                # cannot determine remotely whether it is a file or a
+                # directory).
+                if isfile(self.source):
+                    # dest is not normalized here
+                    arcname = basename(self.dest) or \
+                              basename(normpath(self.source))
+                    destdir = dirname(self.dest)
+                else:
+                    arcname = basename(normpath(self.source))
+                    destdir = os.path.normpath(self.dest)
+                self.logger.debug("copy arcname=%s destdir=%s", arcname,
+                                  destdir)
 
         # And launch stuffs
         next_hops = self._distribute(self.task.info("fanout"), nodes.copy())
@@ -238,15 +263,18 @@ def _launch(self, nodes):
                 self._target_count += len(targets)
                 if self.remote:
                     if self.source:
-                        self.logger.debug('_launch remote untar (destdir=%s)',
-                                          destdir)
-                        self.command = self.UNTAR_CMD_FMT % destdir
-                        worker = self.task.shell(self.command,
-                                                 nodes=targets,
-                                                 timeout=self.timeout,
-                                                 handler=self.metahandler,
-                                                 stderr=self.stderr,
-                                                 tree=False)
+                        # Note: specific case where targets are not in topology
+                        # as self.source is never used on remote gateways
+                        # so we try a direct copy/rcopy:
+                        self.logger.debug('_launch copy r=%s source=%s dest=%s',
+                                          self.reverse, self.source, self.dest)
+                        worker = self.task.copy(self.source, self.dest, targets,
+                                                handler=self.metahandler,
+                                                stderr=self.stderr,
+                                                timeout=self.timeout,
+                                                preserve=self.preserve,
+                                                reverse=self.reverse,
+                                                tree=False)
                     else:
                         worker = self.task.shell(self.command,
                                                  nodes=targets,
@@ -270,13 +298,13 @@ def _launch(self, nodes):
                 self.logger.debug("trying gateway %s to reach %s", gw, targets)
                 if self.source:
                     self._copy_remote(self.source, destdir, targets, gw,
-                                      self.timeout)
+                                      self.timeout, self.reverse)
                 else:
                     self._execute_remote(self.command, targets, gw,
                                          self.timeout)
 
         # Copy mode: send tar data after above workers have been initialized
-        if self.source:
+        if self.source and not self.reverse:
             try:
                 # create temporary tar file with all source files
                 tmptar = tempfile.TemporaryFile()
@@ -305,16 +333,25 @@ def _distribute(self, fanout, dst_nodeset):
                 distribution[gw] = dstset
         return distribution
 
-    def _copy_remote(self, source, dest, targets, gateway, timeout):
+    def _copy_remote(self, source, dest, targets, gateway, timeout, reverse):
         """run a remote copy in tree mode (using gateway)"""
-        self.logger.debug("_copy_remote gateway=%s source=%s dest=%s",
-                          gateway, source, dest)
+        self.logger.debug("_copy_remote gateway=%s source=%s dest=%s "
+                          "reverse=%s", gateway, source, dest, reverse)
 
         self._target_count += len(targets)
 
         self.gwtargets[gateway] = targets.copy()
 
-        cmd = self.UNTAR_CMD_FMT % dest
+        # tar commands are built here and launched on targets
+        if reverse:
+            # these weird replace calls aim to escape single quotes ' within ''
+            srcdir = dirname(source).replace("'", '\'\"\'\"\'')
+            srcbase = basename(normpath(self.source)).replace("'", '\'\"\'\"\'')
+            cmd = self.TAR_CMD_FMT % (srcdir, srcbase)
+        else:
+            cmd = self.UNTAR_CMD_FMT % dest.replace("'", '\'\"\'\"\'')
+
+        self.logger.debug('_copy_remote: tar cmd: %s', cmd)
 
         pchan = self.task._pchannel(gateway, self)
         pchan.shell(nodes=targets, command=cmd, worker=self, timeout=timeout,
@@ -343,17 +380,61 @@ def _engine_clients(self):
         return []
 
     def _on_remote_node_msgline(self, node, msg, sname, gateway):
-        DistantWorker._on_node_msgline(self, node, msg, sname)
+        """remote msg received"""
+        if not self.source or not self.reverse or sname != 'stdout':
+            DistantWorker._on_node_msgline(self, node, msg, sname)
+            return
+
+        # rcopy only: we expect base64 encoded tar content on stdout
+        encoded = self._rcopy_bufs.setdefault(node, '') + msg
+        if node not in self._rcopy_tars:
+            self._rcopy_tars[node] = tempfile.TemporaryFile()
+
+        # partial base64 decoding requires a multiple of 4 characters
+        encoded_sz = (len(encoded) // 4) * 4
+        # write decoded binary msg to node temporary tarfile
+        self._rcopy_tars[node].write(base64.b64decode(encoded[0:encoded_sz]))
+        # keep trailing encoded chars for next time
+        self._rcopy_bufs[node] = encoded[encoded_sz:]
 
     def _on_remote_node_rc(self, node, rc, gateway):
+        """remote rc received"""
         DistantWorker._on_node_rc(self, node, rc)
         self.logger.debug("_on_remote_node_rc %s %s via gw %s", node,
                           self._close_count, gateway)
+
+        # finalize rcopy: extract tar data
+        if self.source and self.reverse:
+            for node, buf in self._rcopy_bufs.iteritems():
+                tarfileobj = self._rcopy_tars[node]
+                if len(buf) > 0:
+                    self.logger.debug("flushing node %s buf %d bytes", node,
+                                      len(buf))
+                    tarfileobj.write(buf)
+                tarfileobj.flush()
+                tarfileobj.seek(0)
+                try:
+                    tmptar = tarfile.open(fileobj=tarfileobj)
+                    try:
+                        self.logger.debug("%s extracting %d members in dest %s",
+                                          node, len(tmptar.getmembers()),
+                                          self.dest)
+                        tmptar.extractall(path=self.dest)
+                    except IOError, ex:
+                        self._on_remote_node_msgline(node, ex, 'stderr',
+                                                     gateway)
+                # note: try-except-finally not supported before python 2.5
+                finally:
+                    tmptar.close()
+            self._rcopy_bufs = {}
+            self._rcopy_tars = {}
+
         self.gwtargets[gateway].remove(node)
         self._close_count += 1
         self._check_fini(gateway)
 
     def _on_remote_node_timeout(self, node, gateway):
+        """remote node timeout received"""
         DistantWorker._on_node_timeout(self, node)
         self.logger.debug("_on_remote_node_timeout %s via gw %s", node, gateway)
         self._close_count += 1