From 59d53e96641e45bc2a8ba1ec31eab653923a7a47 Mon Sep 17 00:00:00 2001 From: Edwin Eefting Date: Tue, 11 May 2021 00:59:26 +0200 Subject: [PATCH] --recv-pipe and --send-pipe implemented. Added CmdItem to make CmdPipe more consitent --- tests/test_cmdpipe.py | 50 +++++++------- zfs_autobackup/CmdPipe.py | 113 +++++++++++++++++++------------- zfs_autobackup/ExecuteNode.py | 36 +++++----- zfs_autobackup/ZfsAutobackup.py | 33 ++++++++-- zfs_autobackup/ZfsDataset.py | 21 +++--- 5 files changed, 151 insertions(+), 102 deletions(-) diff --git a/tests/test_cmdpipe.py b/tests/test_cmdpipe.py index 48ec9ca..855d7ba 100644 --- a/tests/test_cmdpipe.py +++ b/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) diff --git a/zfs_autobackup/CmdPipe.py b/zfs_autobackup/CmdPipe.py index 0853a3b..da1a789 100644 --- a/zfs_autobackup/CmdPipe.py +++ b/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 diff --git a/zfs_autobackup/ExecuteNode.py b/zfs_autobackup/ExecuteNode.py index ca1e965..ae0d8be 100644 --- a/zfs_autobackup/ExecuteNode.py +++ b/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 diff --git a/zfs_autobackup/ZfsAutobackup.py b/zfs_autobackup/ZfsAutobackup.py index b12dee3..35c910b 100644 --- a/zfs_autobackup/ZfsAutobackup.py +++ b/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)) diff --git a/zfs_autobackup/ZfsDataset.py b/zfs_autobackup/ZfsDataset.py index 9f63974..943b51b 100644 --- a/zfs_autobackup/ZfsDataset.py +++ b/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"""