--recv-pipe and --send-pipe implemented. Added CmdItem to make CmdPipe more consitent

tests/test_cmdpipe.py

@@ -1,5 +1,5 @@
 from basetest import *
-from zfs_autobackup.CmdPipe import CmdPipe
+from zfs_autobackup.CmdPipe import CmdPipe,CmdItem
 
 
 class TestCmdPipe(unittest2.TestCase):
@@ -9,24 +9,24 @@ class TestCmdPipe(unittest2.TestCase):
         p=CmdPipe(readonly=False, inp=None)
         err=[]
         out=[]
-        p.add(["ls", "-d", "/", "/", "/nonexistent"], stderr_handler=lambda line: err.append(line), exit_handler=lambda exit_code: self.assertEqual(exit_code,2))
+        p.add(CmdItem(["ls", "-d", "/", "/", "/nonexistent"], stderr_handler=lambda line: err.append(line), exit_handler=lambda exit_code: self.assertEqual(exit_code,2)))
         executed=p.execute(stdout_handler=lambda line: out.append(line))
 
         self.assertEqual(err, ["ls: cannot access '/nonexistent': No such file or directory"])
         self.assertEqual(out, ["/","/"])
-        self.assertTrue(executed)
+        self.assertIsNone(executed)
 
     def test_input(self):
         """test stdinput"""
         p=CmdPipe(readonly=False, inp="test")
         err=[]
         out=[]
-        p.add(["echo", "test"], stderr_handler=lambda line: err.append(line), exit_handler=lambda exit_code: self.assertEqual(exit_code,0))
+        p.add(CmdItem(["echo", "test"], stderr_handler=lambda line: err.append(line), exit_handler=lambda exit_code: self.assertEqual(exit_code,0)))
         executed=p.execute(stdout_handler=lambda line: out.append(line))
 
         self.assertEqual(err, [])
         self.assertEqual(out, ["test"])
-        self.assertTrue(executed)
+        self.assertIsNone(executed)
 
     def test_pipe(self):
         """test piped"""
@@ -35,16 +35,16 @@ class TestCmdPipe(unittest2.TestCase):
         err2=[]
         err3=[]
         out=[]
-        p.add(["echo", "test"], stderr_handler=lambda line: err1.append(line), exit_handler=lambda exit_code: self.assertEqual(exit_code,0))
-        p.add(["tr", "e", "E"], stderr_handler=lambda line: err2.append(line), exit_handler=lambda exit_code: self.assertEqual(exit_code,0))
-        p.add(["tr", "t", "T"], stderr_handler=lambda line: err3.append(line), exit_handler=lambda exit_code: self.assertEqual(exit_code,0))
+        p.add(CmdItem(["echo", "test"], stderr_handler=lambda line: err1.append(line), exit_handler=lambda exit_code: self.assertEqual(exit_code,0)))
+        p.add(CmdItem(["tr", "e", "E"], stderr_handler=lambda line: err2.append(line), exit_handler=lambda exit_code: self.assertEqual(exit_code,0)))
+        p.add(CmdItem(["tr", "t", "T"], stderr_handler=lambda line: err3.append(line), exit_handler=lambda exit_code: self.assertEqual(exit_code,0)))
         executed=p.execute(stdout_handler=lambda line: out.append(line))
 
         self.assertEqual(err1, [])
         self.assertEqual(err2, [])
         self.assertEqual(err3, [])
         self.assertEqual(out, ["TEsT"])
-        self.assertTrue(executed)
+        self.assertIsNone(executed)
 
         #test str representation as well
         self.assertEqual(str(p), "(echo test) | (tr e E) | (tr t T)")
@@ -56,16 +56,16 @@ class TestCmdPipe(unittest2.TestCase):
         err2=[]
         err3=[]
         out=[]
-        p.add(["ls", "/nonexistent1"], stderr_handler=lambda line: err1.append(line), exit_handler=lambda exit_code: self.assertEqual(exit_code,2))
-        p.add(["ls", "/nonexistent2"], stderr_handler=lambda line: err2.append(line), exit_handler=lambda exit_code: self.assertEqual(exit_code,2))
-        p.add(["ls", "/nonexistent3"], stderr_handler=lambda line: err3.append(line), exit_handler=lambda exit_code: self.assertEqual(exit_code,2))
+        p.add(CmdItem(["ls", "/nonexistent1"], stderr_handler=lambda line: err1.append(line), exit_handler=lambda exit_code: self.assertEqual(exit_code,2)))
+        p.add(CmdItem(["ls", "/nonexistent2"], stderr_handler=lambda line: err2.append(line), exit_handler=lambda exit_code: self.assertEqual(exit_code,2)))
+        p.add(CmdItem(["ls", "/nonexistent3"], stderr_handler=lambda line: err3.append(line), exit_handler=lambda exit_code: self.assertEqual(exit_code,2)))
         executed=p.execute(stdout_handler=lambda line: out.append(line))
 
         self.assertEqual(err1, ["ls: cannot access '/nonexistent1': No such file or directory"])
         self.assertEqual(err2, ["ls: cannot access '/nonexistent2': No such file or directory"])
         self.assertEqual(err3, ["ls: cannot access '/nonexistent3': No such file or directory"])
         self.assertEqual(out, [])
-        self.assertTrue(executed)
+        self.assertIsNone(executed)
 
     def test_exitcode(self):
         """test piped exitcodes """
@@ -74,16 +74,16 @@ class TestCmdPipe(unittest2.TestCase):
         err2=[]
         err3=[]
         out=[]
-        p.add(["bash", "-c", "exit 1"], stderr_handler=lambda line: err1.append(line), exit_handler=lambda exit_code: self.assertEqual(exit_code,1))
-        p.add(["bash", "-c", "exit 2"], stderr_handler=lambda line: err2.append(line), exit_handler=lambda exit_code: self.assertEqual(exit_code,2))
-        p.add(["bash", "-c", "exit 3"], stderr_handler=lambda line: err3.append(line), exit_handler=lambda exit_code: self.assertEqual(exit_code,3))
+        p.add(CmdItem(["bash", "-c", "exit 1"], stderr_handler=lambda line: err1.append(line), exit_handler=lambda exit_code: self.assertEqual(exit_code,1)))
+        p.add(CmdItem(["bash", "-c", "exit 2"], stderr_handler=lambda line: err2.append(line), exit_handler=lambda exit_code: self.assertEqual(exit_code,2)))
+        p.add(CmdItem(["bash", "-c", "exit 3"], stderr_handler=lambda line: err3.append(line), exit_handler=lambda exit_code: self.assertEqual(exit_code,3)))
         executed=p.execute(stdout_handler=lambda line: out.append(line))
 
         self.assertEqual(err1, [])
         self.assertEqual(err2, [])
         self.assertEqual(err3, [])
         self.assertEqual(out, [])
-        self.assertTrue(executed)
+        self.assertIsNone(executed)
 
     def test_readonly_execute(self):
         """everything readonly, just should execute"""
@@ -92,16 +92,18 @@ class TestCmdPipe(unittest2.TestCase):
         err1=[]
         err2=[]
         out=[]
-        p.add(["echo", "test1"], stderr_handler=lambda line: err1.append(line), readonly=True)
-        p.add(["echo", "test2"], stderr_handler=lambda line: err2.append(line), readonly=True)
+
+        def true_exit(exit_code):
+            return True
+
+        p.add(CmdItem(["echo", "test1"], stderr_handler=lambda line: err1.append(line), exit_handler=true_exit, readonly=True))
+        p.add(CmdItem(["echo", "test2"], stderr_handler=lambda line: err2.append(line), exit_handler=true_exit, readonly=True))
         executed=p.execute(stdout_handler=lambda line: out.append(line))
 
         self.assertEqual(err1, [])
         self.assertEqual(err2, [])
         self.assertEqual(out, ["test2"])
         self.assertTrue(executed)
-        self.assertEqual(p.items[0]['process'].returncode,0)
-        self.assertEqual(p.items[1]['process'].returncode,0)
 
     def test_readonly_skip(self):
         """one command not readonly, skip"""
@@ -110,12 +112,12 @@ class TestCmdPipe(unittest2.TestCase):
         err1=[]
         err2=[]
         out=[]
-        p.add(["echo", "test1"], stderr_handler=lambda line: err1.append(line), readonly=False)
-        p.add(["echo", "test2"], stderr_handler=lambda line: err2.append(line), readonly=True)
+        p.add(CmdItem(["echo", "test1"], stderr_handler=lambda line: err1.append(line), readonly=False))
+        p.add(CmdItem(["echo", "test2"], stderr_handler=lambda line: err2.append(line), readonly=True))
         executed=p.execute(stdout_handler=lambda line: out.append(line))
 
         self.assertEqual(err1, [])
         self.assertEqual(err2, [])
         self.assertEqual(out, [])
-        self.assertFalse(executed)
+        self.assertTrue(executed)
 

zfs_autobackup/CmdPipe.py

@@ -1,13 +1,50 @@
 import subprocess
 import os
 import select
-import shlex
 
 try:
     from shlex import quote as cmd_quote
 except ImportError:
     from pipes import quote as cmd_quote
 
+
+class CmdItem:
+    """one command item, to be added to a CmdPipe"""
+
+    def __init__(self, cmd, readonly=False, stderr_handler=None, exit_handler=None, shell=False):
+        """create item. caller has to make sure cmd is properly escaped when using shell.
+        :type cmd: list of str
+        """
+
+        self.cmd = cmd
+        self.readonly = readonly
+        self.stderr_handler = stderr_handler
+        self.exit_handler = exit_handler
+        self.shell = shell
+        self.process = None
+
+    def __str__(self):
+        """return copy-pastable version of command."""
+        if self.shell:
+            # its already copy pastable for a shell:
+            return " ".join(self.cmd)
+        else:
+            # make it copy-pastable, will make a mess of quotes sometimes, but is correct
+            return " ".join(map(cmd_quote, self.cmd))
+
+    def create(self, stdin):
+        """actually create the subprocess (called by CmdPipe)"""
+
+        # make sure the command gets all the data in utf8 format:
+        # (this is necessary if LC_ALL=en_US.utf8 is not set in the environment)
+        encoded_cmd = []
+        for arg in self.cmd:
+            encoded_cmd.append(arg.encode('utf-8'))
+
+        self.process = subprocess.Popen(encoded_cmd, env=os.environ, stdout=subprocess.PIPE, stdin=stdin,
+                                        stderr=subprocess.PIPE, shell=self.shell)
+
+
 class CmdPipe:
     """a pipe of one or more commands. also takes care of utf-8 encoding/decoding and line based parsing"""
 
@@ -23,43 +60,35 @@ class CmdPipe:
         self.readonly = readonly
         self._should_execute = True
 
-    def add(self, cmd, readonly=False, stderr_handler=None, exit_handler=None, shell=False):
-        """adds a command to pipe. called has to make sure its properly escaped."""
+    def add(self, cmd_item):
+        """adds a CmdItem to pipe.
+        :type cmd_item: CmdItem
+        """
 
-        self.items.append({
-            'cmd': cmd,
-            'stderr_handler': stderr_handler,
-            'exit_handler': exit_handler,
-            'shell': shell
-        })
+        self.items.append(cmd_item)
 
-        if not readonly and self.readonly:
+        if not cmd_item.readonly and self.readonly:
             self._should_execute = False
 
     def __str__(self):
-        """transform into oneliner for debugging and testing. this should generate a copy-pastable string for in a console """
+        """transform whole pipe into oneliner for debugging and testing. this should generate a copy-pastable string for in a console """
 
         ret = ""
         for item in self.items:
             if ret:
                 ret = ret + " | "
-            if item['shell']:
-                #its already copy pastable for a shell:
-                ret = ret + "(" + " ".join(item['cmd']) + ")"
-            else:
-                #make it copy-pastable, will make a mess of quotes sometimes, but is correct
-                ret = ret + "(" + " ".join(map(cmd_quote,item['cmd'])) + ")"
+            ret = ret + "({})".format(item)  # this will do proper escaping to make it copypastable
 
         return ret
 
     def should_execute(self):
-        return(self._should_execute)
+        return self._should_execute
 
     def execute(self, stdout_handler):
-        """run the pipe. returns True if it executed, and false if it skipped due to readonly conditions"""
+        """run the pipe. returns True all exit handlers returned true"""
 
         if not self._should_execute:
-            return False
+            return True
 
         # first process should have actual user input as stdin:
         selectors = []
@@ -69,29 +98,21 @@ class CmdPipe:
         stdin = subprocess.PIPE
         for item in self.items:
 
-            # make sure the command gets all the data in utf8 format:
-            # (this is necessary if LC_ALL=en_US.utf8 is not set in the environment)
-            encoded_cmd = []
-            for arg in item['cmd']:
-                encoded_cmd.append(arg.encode('utf-8'))
-
-            item['process'] = subprocess.Popen(encoded_cmd, env=os.environ, stdout=subprocess.PIPE, stdin=stdin,
-                                               stderr=subprocess.PIPE, shell=item['shell'])
-
-            selectors.append(item['process'].stderr)
+            item.create(stdin)
+            selectors.append(item.process.stderr)
 
             if last_stdout is None:
                 # we're the first process in the pipe, do we have some input?
                 if self.inp is not None:
                     # TODO: make streaming to support big inputs?
-                    item['process'].stdin.write(self.inp.encode('utf-8'))
-                item['process'].stdin.close()
+                    item.process.stdin.write(self.inp.encode('utf-8'))
+                item.process.stdin.close()
             else:
-                #last stdout was piped to this stdin already, so close it because we dont need it anymore
+                # last stdout was piped to this stdin already, so close it because we dont need it anymore
                 last_stdout.close()
 
-            last_stdout = item['process'].stdout
-            stdin=last_stdout
+            last_stdout = item.process.stdout
+            stdin = last_stdout
 
         # monitor last stdout as well
         selectors.append(last_stdout)
@@ -111,29 +132,29 @@ class CmdPipe:
                     eof_count = eof_count + 1
 
             for item in self.items:
-                if item['process'].stderr in read_ready:
-                    line = item['process'].stderr.readline().decode('utf-8').rstrip()
+                if item.process.stderr in read_ready:
+                    line = item.process.stderr.readline().decode('utf-8').rstrip()
                     if line != "":
-                        item['stderr_handler'](line)
+                        item.stderr_handler(line)
                     else:
                         eof_count = eof_count + 1
 
-                if item['process'].poll() is not None:
+                if item.process.poll() is not None:
                     done_count = done_count + 1
 
             # all filehandles are eof and all processes are done (poll() is not None)
             if eof_count == len(selectors) and done_count == len(self.items):
                 break
 
-        #close filehandles
+        # close filehandles
         last_stdout.close()
         for item in self.items:
-            item['process'].stderr.close()
+            item.process.stderr.close()
 
-        #call exit handlers
+        # call exit handlers
+        success = True
         for item in self.items:
-            if item['exit_handler'] is not None:
-                item['exit_handler'](item['process'].returncode)
+            if item.exit_handler is not None:
+                success=item.exit_handler(item.process.returncode) and success
 
-
-        return True
+        return success

zfs_autobackup/ExecuteNode.py

@@ -1,7 +1,7 @@
 import os
 import select
 import subprocess
-from zfs_autobackup.CmdPipe import CmdPipe
+from zfs_autobackup.CmdPipe import CmdPipe, CmdItem
 from zfs_autobackup.LogStub import LogStub
 
 try:
@@ -49,14 +49,14 @@ class ExecuteNode(LogStub):
         else:
             self.error("STDERR > " + line.rstrip())
 
-    def __quote(self, cmd):
+    def _quote(self, cmd):
         """return quoted version of command. if it has value PIPE it will add an actual | """
         if cmd==self.PIPE:
             return('|')
         else:
             return(cmd_quote(cmd))
 
-    def __shell_cmd(self, cmd):
+    def _shell_cmd(self, cmd):
         """prefix specified ssh shell to command and escape shell characters"""
 
         ret=[]
@@ -70,14 +70,13 @@ class ExecuteNode(LogStub):
 
             ret.append(self.ssh_to)
 
-        ret.append(" ".join(map(self.__quote, cmd)))
+        ret.append(" ".join(map(self._quote, cmd)))
 
         return ret
 
     def is_local(self):
         return self.ssh_to is None
 
-
     def run(self, cmd, inp=None, tab_split=False, valid_exitcodes=None, readonly=False, hide_errors=False,
             return_stderr=False, pipe=False):
         """run a command on the node , checks output and parses/handle output and returns it
@@ -100,13 +99,14 @@ class ExecuteNode(LogStub):
 
         # create new pipe?
         if not isinstance(inp, CmdPipe):
-            p = CmdPipe(self.readonly, inp)
+            cmd_pipe = CmdPipe(self.readonly, inp)
         else:
             # add stuff to existing pipe
-            p = inp
+            cmd_pipe = inp
 
         # stderr parser
         error_lines = []
+
         def stderr_handler(line):
             if tab_split:
                 error_lines.append(line.rstrip().split('\t'))
@@ -123,17 +123,22 @@ class ExecuteNode(LogStub):
                 self.debug("EXIT   > {}".format(exit_code))
 
             if (valid_exitcodes != []) and (exit_code not in valid_exitcodes):
-             raise (ExecuteError("Command '{}' returned exit code {} (valid codes: {})".format(" ".join(cmd), exit_code, valid_exitcodes)))
+                self.error("Command \"{}\" returned exit code {} (valid codes: {})".format(cmd_item, exit_code, valid_exitcodes))
+                return False
 
-        #add shell command and handlers to pipe
-        p.add(cmd=self.__shell_cmd(cmd), readonly=readonly, stderr_handler=stderr_handler, exit_handler=exit_handler, shell=self.is_local())
+            return True
+
+        # add shell command and handlers to pipe
+        cmd_item=CmdItem(cmd=self._shell_cmd(cmd), readonly=readonly, stderr_handler=stderr_handler, exit_handler=exit_handler, shell=self.is_local())
+        cmd_pipe.add(cmd_item)
 
         # return pipe instead of executing?
         if pipe:
-            return p
+            return cmd_pipe
 
         # stdout parser
         output_lines = []
+
         def stdout_handler(line):
             if tab_split:
                 output_lines.append(line.rstrip().split('\t'))
@@ -141,13 +146,14 @@ class ExecuteNode(LogStub):
                 output_lines.append(line.rstrip())
             self._parse_stdout(line)
 
-        if p.should_execute():
-            self.debug("CMD    > {}".format(p))
+        if cmd_pipe.should_execute():
+            self.debug("CMD    > {}".format(cmd_pipe))
         else:
-            self.debug("CMDSKIP> {}".format(p))
+            self.debug("CMDSKIP> {}".format(cmd_pipe))
 
         # execute and calls handlers in CmdPipe
-        p.execute(stdout_handler=stdout_handler)
+        if not cmd_pipe.execute(stdout_handler=stdout_handler):
+            raise(ExecuteError("Last command returned error"))
 
         if return_stderr:
             return output_lines, error_lines

zfs_autobackup/ZfsAutobackup.py

@@ -2,6 +2,7 @@ import argparse
 import sys
 import time
 
+from zfs_autobackup.ExecuteNode import ExecuteNode
 from zfs_autobackup.Thinner import Thinner
 from zfs_autobackup.ZfsDataset import ZfsDataset
 from zfs_autobackup.LogConsole import LogConsole
@@ -9,6 +10,7 @@ from zfs_autobackup.ZfsNode import ZfsNode
 from zfs_autobackup.ThinnerRule import ThinnerRule
 
 
+
 class ZfsAutobackup:
     """main class"""
 
@@ -109,10 +111,10 @@ class ZfsAutobackup:
                             help=argparse.SUPPRESS)  # needed to workaround a zfs recv -v bug
 
         parser.add_argument('--send-pipe', metavar="COMMAND", default=[], action='append',
-                            help='pipe zfs send output through COMMAND')
+                            help='pipe zfs send output through COMMAND (can be used multiple times)')
 
         parser.add_argument('--recv-pipe', metavar="COMMAND", default=[], action='append',
-                            help='pipe zfs recv input through COMMAND')
+                            help='pipe zfs recv input through COMMAND (can be used multiple times)')
 
         parser.add_argument('--resume', action='store_true', help=argparse.SUPPRESS)
         parser.add_argument('--raw', action='store_true', help=argparse.SUPPRESS)
@@ -259,6 +261,26 @@ class ZfsAutobackup:
         if self.args.progress:
             self.clear_progress()
 
+    def get_input_pipes(self):
+
+        ret=[]
+
+        for input_pipe in self.args.recv_pipe:
+            ret.extend(input_pipe.split(" "))
+            ret.append(ExecuteNode.PIPE)
+
+        return ret
+
+    def get_output_pipes(self):
+
+        ret=[]
+
+        for output_pipe in self.args.send_pipe:
+            ret.append(ExecuteNode.PIPE)
+            ret.extend(output_pipe.split(" "))
+
+        return ret
+
     # NOTE: this method also uses self.args. args that need extra processing are passed as function parameters:
     def sync_datasets(self, source_node, source_datasets, target_node):
         """Sync datasets, or thin-only on both sides
@@ -267,6 +289,9 @@ class ZfsAutobackup:
         :type source_node: ZfsNode
         """
 
+        output_pipes=self.get_output_pipes()
+        input_pipes=self.get_input_pipes()
+
         fail_count = 0
         count = 0
         target_datasets = []
@@ -304,8 +329,8 @@ class ZfsAutobackup:
                                               also_other_snapshots=self.args.other_snapshots,
                                               no_send=self.args.no_send,
                                               destroy_incompatible=self.args.destroy_incompatible,
-                                              output_pipes=self.args.send_pipe, input_pipes=self.args.recv_pipe,
-                                              decrypt=self.args.decrypt, encrypt=self.args.encrypt)
+                                              output_pipes=output_pipes, input_pipes=input_pipes,
+                                              decrypt=self.args.decrypt, encrypt=self.args.encrypt, )
             except Exception as e:
                 fail_count = fail_count + 1
                 source_dataset.error("FAILED: " + str(e))

zfs_autobackup/ZfsDataset.py

@@ -510,6 +510,7 @@ class ZfsDataset:
         need to know snapshot names)
 
         Args:
+            :param output_pipes: output cmd array that will be added to actual zfs send command. (e.g. mbuffer or compression program)
             :type output_pipes: list of str
             :type features: list of str
             :type prev_snapshot: ZfsDataset
@@ -556,22 +557,13 @@ class ZfsDataset:
 
             cmd.append(self.name)
 
-        # #add custom output pipes?
-        # #local so do our own piping
-        # if self.zfs_node.is_local():
-        #     output_pipe = self.zfs_node.run(cmd, pipe=True, readonly=True)
-        #     for pipe_cmd in output_pipes:
-        #         output_pipe=self.zfs_node.run(pipe_cmd.split(" "), inp=output_pipe, pipe=True, readonly=False)
-        # #remote, so add with actual | and let remote shell handle it
-        # else:
-        #     for pipe_cmd in output_pipes:
-        #         cmd.append("|")
-        #         cmd.extend(pipe_cmd.split(" "))
+        cmd.extend(output_pipes)
+
         output_pipe = self.zfs_node.run(cmd, pipe=True, readonly=True)
 
         return output_pipe
 
-    def recv_pipe(self, pipe, features, filter_properties=None, set_properties=None, ignore_exit_code=False):
+    def recv_pipe(self, pipe, features, input_pipes, filter_properties=None, set_properties=None, ignore_exit_code=False):
         """starts a zfs recv for this snapshot and uses pipe as input
 
         note: you can it both on a snapshot or filesystem object. The
@@ -579,6 +571,7 @@ class ZfsDataset:
         differently.
 
         Args:
+            :param input_pipes: input cmd array that will be prepended to actual zfs recv command. (e.g. mbuffer or decompression program)
             :type pipe: subprocess.pOpen
             :type features: list of str
             :type filter_properties: list of str
@@ -595,6 +588,8 @@ class ZfsDataset:
         # build target command
         cmd = []
 
+        cmd.extend(input_pipes)
+
         cmd.extend(["zfs", "recv"])
 
         # don't mount filesystem that is received
@@ -680,7 +675,7 @@ class ZfsDataset:
         pipe = self.send_pipe(features=features, show_progress=show_progress, prev_snapshot=prev_snapshot,
                               resume_token=resume_token, raw=raw, send_properties=send_properties, write_embedded=write_embedded, output_pipes=output_pipes)
         target_snapshot.recv_pipe(pipe, features=features, filter_properties=filter_properties,
-                                  set_properties=set_properties, ignore_exit_code=ignore_recv_exit_code)
+                                  set_properties=set_properties, ignore_exit_code=ignore_recv_exit_code, input_pipes=input_pipes)
 
     def abort_resume(self):
         """abort current resume state"""