fix #84

Add support for pre- and post-snapshot scripts (#39)

README.md

@@ -400,7 +400,7 @@ I'll add some tips when the issues start to get in on github. :)
 
 ## Transfer buffering, compression and rate limiting.
 
-If you're transferring over a slow link it might be useful to use `--compress=zstd-fast`. This will compres the data before sending, so it uses less bandwidth.
+If you're transferring over a slow link it might be useful to use `--compress=zstd-fast`. This will compress the data before sending, so it uses less bandwidth. An alternative to this is to use --zfs-compressed: This will transfer blocks that already have compression intact. (--compress will usually compress much better but uses much more resources. --zfs-compressed uses the least resources, but can be a disadvantage if you want to use a different compression method on the target.)
 
 You can also limit the datarate by using the `--rate` option.
 
@@ -417,6 +417,28 @@ zfs send -> send buffer -> custom send pipes -> compression -> transfer rate lim
 #### On the receiving side:
 decompression -> custom recv pipes -> buffer -> zfs recv
 
+## Running custom commands before and after snapshotting
+
+You can run commands before and after the snapshot to freeze databases to make the on for example to make the on-disk data consistent before snapshotting.
+
+The commands will be executed on the source side. Use the `--pre-snapshot-cmd` and `--post-snapshot-cmd` options for this.
+
+For example:
+
+```sh
+zfs-autobackup \
+    --pre-snapshot-cmd 'daemon -f jexec mysqljail1 mysql -s -e "set autocommit=0;flush logs;flush tables with read lock;\\! echo \$\$ > /tmp/mysql_lock.pid && sleep 60"' \
+    --pre-snapshot-cmd 'daemon -f jexec mysqljail2 mysql -s -e "set autocommit=0;flush logs;flush tables with read lock;\\! echo \$\$ > /tmp/mysql_lock.pid && sleep 60"' \
+    --post-snapshot-cmd 'pkill -F /jails/mysqljail1/tmp/mysql_lock.pid' \
+    --post-snapshot-cmd 'pkill -F /jails/mysqljail2/tmp/mysql_lock.pid' \
+    backupfs1
+```
+
+Failure handling during pre/post commands:
+
+* If a pre-command fails, zfs-autobackup will exit with an error. (after executing the post-commands)
+* All post-commands are always executed. Even if the pre-commands or actual snapshot have failed. This way you can be sure that stuff is always cleanedup and unfreezed.
+
 ## Tips
 
 * Use ```--debug``` if something goes wrong and you want to see the commands that are executed. This will also stop at the first error.
@@ -493,7 +515,8 @@ Look in man ssh_config for many more options.
 ```console
 usage: zfs-autobackup [-h] [--ssh-config CONFIG-FILE] [--ssh-source USER@HOST]
                       [--ssh-target USER@HOST] [--keep-source SCHEDULE]
-                      [--keep-target SCHEDULE] [--other-snapshots]
+                      [--keep-target SCHEDULE] [--pre-snapshot-cmd COMMAND]
+                      [--post-snapshot-cmd COMMAND] [--other-snapshots]
                       [--no-snapshot] [--no-send] [--no-thinning] [--no-holds]
                       [--min-change BYTES] [--allow-empty]
                       [--ignore-replicated] [--strip-path N]
@@ -531,6 +554,12 @@ optional arguments:
   --keep-target SCHEDULE
                         Thinning schedule for old target snapshots. Default:
                         10,1d1w,1w1m,1m1y
+  --pre-snapshot-cmd COMMAND
+                        Run COMMAND before snapshotting (can be used multiple
+                        times.
+  --post-snapshot-cmd COMMAND
+                        Run COMMAND after snapshotting (can be used multiple
+                        times.
   --other-snapshots     Send over other snapshots as well, not just the ones
                         created by this tool.
   --no-snapshot         Don't create new snapshots (useful for finishing

tests/test_externalfailures.py

@@ -227,11 +227,11 @@ test_target1/test_source2/fs2/sub@test-20101111000000
                 # incremental, doesnt want previous anymore
                 with patch('time.strftime', return_value="20101111000002"):
                     self.assertFalse(ZfsAutobackup(
-                        "test test_target1 --no-progress --verbose --keep-target=0 --debug --allow-empty".split(" ")).run())
+                        "test test_target1 --no-progress --verbose --keep-target=0 --allow-empty".split(" ")).run())
 
             print(buf.getvalue())
 
-            self.assertIn(": aborting resume, since", buf.getvalue())
+            self.assertIn("Aborting resume, we dont want that snapshot anymore.", buf.getvalue())
 
         r = shelltest("zfs list -H -o name -r -t all test_target1")
         self.assertMultiLineEqual(r, """
@@ -247,6 +247,34 @@ test_target1/test_source2/fs2/sub
 test_target1/test_source2/fs2/sub@test-20101111000002
 """)
 
+    # test with empty snapshot list (this was a bug)
+    def test_abort_resume_emptysnapshotlist(self):
+
+        if "0.6.5" in ZFS_USERSPACE:
+            self.skipTest("Resume not supported in this ZFS userspace version")
+
+        with patch('time.strftime', return_value="20101111000000"):
+            self.assertFalse(ZfsAutobackup("test test_target1 --no-progress --verbose".split(" ")).run())
+
+        # generate resume
+        with patch('time.strftime', return_value="20101111000001"):
+            self.generate_resume()
+
+        shelltest("zfs destroy test_source1/fs1@test-20101111000001")
+
+        with OutputIO() as buf:
+            with redirect_stdout(buf):
+                # incremental, doesnt want previous anymore
+                with patch('time.strftime', return_value="20101111000002"):
+                    self.assertFalse(ZfsAutobackup(
+                        "test test_target1 --no-progress --verbose --no-snapshot".split(
+                            " ")).run())
+
+            print(buf.getvalue())
+
+            self.assertIn("Aborting resume, its obsolete", buf.getvalue())
+
+
     def test_missing_common(self):
 
         with patch('time.strftime', return_value="20101111000000"):

tests/test_regressions.py

@@ -8,12 +8,51 @@ class TestZfsNode(unittest2.TestCase):
         prepare_zpools()
         self.longMessage=True
 
-    # #resume initial backup
-    # def test_keepsource0(self):
+    def test_keepsource0target10queuedsend(self):
+        """Test if thinner doesnt destroy too much early on if there are no common snapshots YET. Issue #84"""
 
-    #     #somehow only specifying --allow-empty --keep-source 0 failed:
-    #     with patch('time.strftime', return_value="20101111000000"):
-    #         self.assertFalse(ZfsAutobackup("test test_target1 --verbose --allow-empty --keep-source 0".split(" ")).run())
+        with patch('time.strftime', return_value="20101111000000"):
+            self.assertFalse(ZfsAutobackup(
+                "test test_target1 --no-progress --verbose --keep-source=0 --keep-target=10 --allow-empty --no-send".split(
+                    " ")).run())
 
-    #     with patch('time.strftime', return_value="20101111000001"):
-    #         self.assertFalse(ZfsAutobackup("test test_target1 --verbose --allow-empty --keep-source 0".split(" ")).run())
+        with patch('time.strftime', return_value="20101111000001"):
+            self.assertFalse(ZfsAutobackup(
+                "test test_target1 --no-progress --verbose --keep-source=0 --keep-target=10 --allow-empty --no-send".split(
+                    " ")).run())
+
+        with patch('time.strftime', return_value="20101111000002"):
+            self.assertFalse(ZfsAutobackup(
+                "test test_target1 --no-progress --verbose --keep-source=0 --keep-target=10 --allow-empty".split(
+                    " ")).run())
+
+        r = shelltest("zfs list -H -o name -r -t all " + TEST_POOLS)
+        self.assertMultiLineEqual(r, """
+test_source1
+test_source1/fs1
+test_source1/fs1@test-20101111000002
+test_source1/fs1/sub
+test_source1/fs1/sub@test-20101111000002
+test_source2
+test_source2/fs2
+test_source2/fs2/sub
+test_source2/fs2/sub@test-20101111000002
+test_source2/fs3
+test_source2/fs3/sub
+test_target1
+test_target1/test_source1
+test_target1/test_source1/fs1
+test_target1/test_source1/fs1@test-20101111000000
+test_target1/test_source1/fs1@test-20101111000001
+test_target1/test_source1/fs1@test-20101111000002
+test_target1/test_source1/fs1/sub
+test_target1/test_source1/fs1/sub@test-20101111000000
+test_target1/test_source1/fs1/sub@test-20101111000001
+test_target1/test_source1/fs1/sub@test-20101111000002
+test_target1/test_source2
+test_target1/test_source2/fs2
+test_target1/test_source2/fs2/sub
+test_target1/test_source2/fs2/sub@test-20101111000000
+test_target1/test_source2/fs2/sub@test-20101111000001
+test_target1/test_source2/fs2/sub@test-20101111000002
+""")

tests/test_zfsautobackup.py

@@ -312,8 +312,6 @@ test_target1/test_source2/fs2
             self.assertFalse(ZfsAutobackup("test test_target1 --verbose --no-send --no-progress".split(" ")).run())
 
             r=shelltest("zfs list -H -o name -r -t all "+TEST_POOLS)
-            #(only parents are created )
-            #TODO: it probably shouldn't create these
             self.assertMultiLineEqual(r,"""
 test_source1
 test_source1/fs1
@@ -337,8 +335,6 @@ test_target1
             self.assertFalse(ZfsAutobackup("test test_target1 --no-progress --verbose --ignore-replicated".split(" ")).run())
 
             r=shelltest("zfs list -H -o name -r -t all "+TEST_POOLS)
-            #(only parents are created )
-            #TODO: it probably shouldn't create these
             self.assertMultiLineEqual(r,"""
 test_source1
 test_source1/fs1
@@ -851,7 +847,7 @@ test_target1/test_source2/fs2/sub@test-20101111000003
 """)
 
 
-        # make snapshot 4, since we used no-holds, it will delete 3 on the source, breaking the backup
+        # run with snapshot-only for 4, since we used no-holds, it will delete 3 on the source, breaking the backup
         with patch('time.strftime', return_value="20101111000004"):
             self.assertFalse(ZfsAutobackup("test --no-progress --verbose --keep-source=0 --keep-target=0 --allow-empty".split(" ")).run())
 
@@ -890,7 +886,7 @@ test_target1/test_source2/fs2/sub@test-20101111000003
         n=ZfsNode("test",l)
         d=ZfsDataset(n,"test_source1@test")
 
-        sp=d.send_pipe([], prev_snapshot=None, resume_token=None, show_progress=True, raw=False, send_pipes=[], send_properties=True, write_embedded=True)
+        sp=d.send_pipe([], prev_snapshot=None, resume_token=None, show_progress=True, raw=False, send_pipes=[], send_properties=True, write_embedded=True, zfs_compressed=True)
 
 
         with OutputIO() as buf:

tests/test_zfsautobackup31.py

@@ -71,4 +71,11 @@ test_target1/b/test_source1/fs1/sub@test-20101111000000
 test_target1/b/test_source2/fs2/sub@test-20101111000000
 test_target1/b/test_target1/a/test_source1/fs1@test-20101111000000
 test_target1/b/test_target1/a/test_source1/fs1/sub@test-20101111000000
-""")
+""")
+
+    def test_zfs_compressed(self):
+
+        with patch('time.strftime', return_value="20101111000000"):
+            self.assertFalse(
+                ZfsAutobackup("test test_target1 --no-progress --verbose --debug --zfs-compressed".split(" ")).run())
+

tests/test_zfsnode.py

@@ -1,6 +1,6 @@
 from basetest import *
 from zfs_autobackup.LogStub import LogStub
-
+from zfs_autobackup.ExecuteNode import ExecuteError
 
 
 class TestZfsNode(unittest2.TestCase):
@@ -9,16 +9,15 @@ class TestZfsNode(unittest2.TestCase):
         prepare_zpools()
         # return super().setUp()
 
-
     def test_consistent_snapshot(self):
-        logger=LogStub()
-        description="[Source]"
-        node=ZfsNode("test", logger, description=description)
+        logger = LogStub()
+        description = "[Source]"
+        node = ZfsNode("test", logger, description=description)
 
         with self.subTest("first snapshot"):
-            node.consistent_snapshot(node.selected_datasets(exclude_paths=[], exclude_received=False), "test-1",100000)
-            r=shelltest("zfs list -H -o name -r -t all "+TEST_POOLS)
-            self.assertEqual(r,"""
+            node.consistent_snapshot(node.selected_datasets(exclude_paths=[], exclude_received=False), "test-1", 100000)
+            r = shelltest("zfs list -H -o name -r -t all " + TEST_POOLS)
+            self.assertEqual(r, """
 test_source1
 test_source1/fs1
 test_source1/fs1@test-1
@@ -33,11 +32,10 @@ test_source2/fs3/sub
 test_target1
 """)
 
-
         with self.subTest("second snapshot, no changes, no snapshot"):
-            node.consistent_snapshot(node.selected_datasets(exclude_paths=[], exclude_received=False), "test-2",1)
-            r=shelltest("zfs list -H -o name -r -t all "+TEST_POOLS)
-            self.assertEqual(r,"""
+            node.consistent_snapshot(node.selected_datasets(exclude_paths=[], exclude_received=False), "test-2", 1)
+            r = shelltest("zfs list -H -o name -r -t all " + TEST_POOLS)
+            self.assertEqual(r, """
 test_source1
 test_source1/fs1
 test_source1/fs1@test-1
@@ -53,9 +51,9 @@ test_target1
 """)
 
         with self.subTest("second snapshot, no changes, empty snapshot"):
-            node.consistent_snapshot(node.selected_datasets(exclude_paths=[], exclude_received=False), "test-2",0)
-            r=shelltest("zfs list -H -o name -r -t all "+TEST_POOLS)
-            self.assertEqual(r,"""
+            node.consistent_snapshot(node.selected_datasets(exclude_paths=[], exclude_received=False), "test-2", 0)
+            r = shelltest("zfs list -H -o name -r -t all " + TEST_POOLS)
+            self.assertEqual(r, """
 test_source1
 test_source1/fs1
 test_source1/fs1@test-1
@@ -73,31 +71,82 @@ test_source2/fs3/sub
 test_target1
 """)
 
+    def test_consistent_snapshot_prepostcmds(self):
+        logger = LogStub()
+        description = "[Source]"
+        node = ZfsNode("test", logger, description=description, debug_output=True)
+
+        with self.subTest("Test if all cmds are executed correctly (no failures)"):
+            with OutputIO() as buf:
+                with redirect_stdout(buf):
+                    node.consistent_snapshot(node.selected_datasets(exclude_paths=[], exclude_received=False), "test-1",
+                                             0,
+                                             pre_snapshot_cmds=["echo pre1", "echo pre2"],
+                                             post_snapshot_cmds=["echo post1 >&2", "echo post2 >&2"]
+                                             )
+
+                self.assertIn("STDOUT > pre1", buf.getvalue())
+                self.assertIn("STDOUT > pre2", buf.getvalue())
+                self.assertIn("STDOUT > post1", buf.getvalue())
+                self.assertIn("STDOUT > post2", buf.getvalue())
+
+
+        with self.subTest("Failure in the middle, only pre1 and both post1 and post2 should be executed, no snapshot should be attempted"):
+            with OutputIO() as buf:
+                with redirect_stdout(buf):
+                    with self.assertRaises(ExecuteError):
+                        node.consistent_snapshot(node.selected_datasets(exclude_paths=[], exclude_received=False), "test-1",
+                                                 0,
+                                                 pre_snapshot_cmds=["echo pre1", "false", "echo pre2"],
+                                                 post_snapshot_cmds=["echo post1", "false", "echo post2"]
+                                                 )
+
+                print(buf.getvalue())
+                self.assertIn("STDOUT > pre1", buf.getvalue())
+                self.assertNotIn("STDOUT > pre2", buf.getvalue())
+                self.assertIn("STDOUT > post1", buf.getvalue())
+                self.assertIn("STDOUT > post2", buf.getvalue())
+
+        with self.subTest("Snapshot fails"):
+            with OutputIO() as buf:
+                with redirect_stdout(buf):
+                    with self.assertRaises(ExecuteError):
+                        #same snapshot name as before so it fails
+                        node.consistent_snapshot(node.selected_datasets(exclude_paths=[], exclude_received=False), "test-1",
+                                                 0,
+                                                 pre_snapshot_cmds=["echo pre1", "echo pre2"],
+                                                 post_snapshot_cmds=["echo post1", "echo post2"]
+                                                 )
+
+                print(buf.getvalue())
+                self.assertIn("STDOUT > pre1", buf.getvalue())
+                self.assertIn("STDOUT > pre2", buf.getvalue())
+                self.assertIn("STDOUT > post1", buf.getvalue())
+                self.assertIn("STDOUT > post2", buf.getvalue())
+
 
     def test_getselected(self):
-        logger=LogStub()
-        description="[Source]"
-        node=ZfsNode("test", logger, description=description)
-        s=pformat(node.selected_datasets(exclude_paths=[], exclude_received=False))
+        logger = LogStub()
+        description = "[Source]"
+        node = ZfsNode("test", logger, description=description)
+        s = pformat(node.selected_datasets(exclude_paths=[], exclude_received=False))
         print(s)
 
-        #basics
-        self.assertEqual (s, """[(local): test_source1/fs1,
+        # basics
+        self.assertEqual(s, """[(local): test_source1/fs1,
  (local): test_source1/fs1/sub,
  (local): test_source2/fs2/sub]""")
 
-        #caching, so expect same result after changing it
+        # caching, so expect same result after changing it
         subprocess.check_call("zfs set autobackup:test=true test_source2/fs3", shell=True)
-        self.assertEqual (s, """[(local): test_source1/fs1,
+        self.assertEqual(s, """[(local): test_source1/fs1,
  (local): test_source1/fs1/sub,
  (local): test_source2/fs2/sub]""")
 
-
     def test_validcommand(self):
-        logger=LogStub()
-        description="[Source]"
-        node=ZfsNode("test", logger, description=description)
-
+        logger = LogStub()
+        description = "[Source]"
+        node = ZfsNode("test", logger, description=description)
 
         with self.subTest("test invalid option"):
             self.assertFalse(node.valid_command(["zfs", "send", "--invalid-option", "nonexisting"]))
@@ -105,21 +154,19 @@ test_target1
             self.assertTrue(node.valid_command(["zfs", "send", "-v", "nonexisting"]))
 
     def test_supportedsendoptions(self):
-        logger=LogStub()
-        description="[Source]"
-        node=ZfsNode("test", logger, description=description)
+        logger = LogStub()
+        description = "[Source]"
+        node = ZfsNode("test", logger, description=description)
         # -D propably always supported
-        self.assertGreater(len(node.supported_send_options),0)
-
+        self.assertGreater(len(node.supported_send_options), 0)
 
     def test_supportedrecvoptions(self):
-        logger=LogStub()
-        description="[Source]"
-        #NOTE: this could hang via ssh if we dont close filehandles properly. (which was a previous bug)
-        node=ZfsNode("test", logger, description=description, ssh_to='localhost')
+        logger = LogStub()
+        description = "[Source]"
+        # NOTE: this could hang via ssh if we dont close filehandles properly. (which was a previous bug)
+        node = ZfsNode("test", logger, description=description, ssh_to='localhost')
         self.assertIsInstance(node.supported_recv_options, list)
 
 
-
 if __name__ == '__main__':
     unittest.main()

zfs_autobackup/ZfsAutobackup.py

@@ -16,7 +16,7 @@ from zfs_autobackup.ThinnerRule import ThinnerRule
 class ZfsAutobackup:
     """main class"""
 
-    VERSION = "3.1-rc1"
+    VERSION = "3.1-rc5"
     HEADER = "zfs-autobackup v{} - (c)2021 E.H.Eefting (edwin@datux.nl)".format(VERSION)
 
     def __init__(self, argv, print_arguments=True):
@@ -45,6 +45,10 @@ class ZfsAutobackup:
                             help='Target ZFS filesystem (optional: if not specified, zfs-autobackup will only operate '
                                  'as snapshot-tool on source)')
 
+        parser.add_argument('--pre-snapshot-cmd', metavar="COMMAND", default=[], action='append',
+                            help='Run COMMAND before snapshotting (can be used multiple times.')
+        parser.add_argument('--post-snapshot-cmd', metavar="COMMAND", default=[], action='append',
+                            help='Run COMMAND after snapshotting (can be used multiple times.')
         parser.add_argument('--other-snapshots', action='store_true',
                             help='Send over other snapshots as well, not just the ones created by this tool.')
         parser.add_argument('--no-snapshot', action='store_true',
@@ -75,7 +79,7 @@ class ZfsAutobackup:
         parser.add_argument('--clear-mountpoint', action='store_true',
                             help='Set property canmount=noauto for new datasets. (recommended, prevents mount '
                                  'conflicts. same as --set-properties canmount=noauto)')
-        parser.add_argument('--filter-properties', metavar='PROPERY,...', type=str,
+        parser.add_argument('--filter-properties', metavar='PROPERTY,...', type=str,
                             help='List of properties to "filter" when receiving filesystems. (you can still restore '
                                  'them with zfs inherit -S)')
         parser.add_argument('--set-properties', metavar='PROPERTY=VALUE,...', type=str,
@@ -99,6 +103,9 @@ class ZfsAutobackup:
         parser.add_argument('--encrypt', action='store_true',
                             help='Encrypt data after receiving it.')
 
+        parser.add_argument('--zfs-compressed', action='store_true',
+                            help='Transfer blocks that already have zfs-compression as-is.')
+
         parser.add_argument('--test', action='store_true',
                             help='dont change anything, just show what would be done (still does all read-only '
                                  'operations)')
@@ -166,6 +173,9 @@ class ZfsAutobackup:
         if args.compress and args.ssh_source is None and args.ssh_target is None:
             self.warning("Using compression, but transfer is local.")
 
+        if args.compress and args.zfs_compressed:
+            self.warning("Using --compress with --zfs-compressed, might be inefficient.")
+
     def verbose(self, txt):
         self.log.verbose(txt)
 
@@ -381,7 +391,7 @@ class ZfsAutobackup:
                                               no_send=self.args.no_send,
                                               destroy_incompatible=self.args.destroy_incompatible,
                                               send_pipes=send_pipes, recv_pipes=recv_pipes,
-                                              decrypt=self.args.decrypt, encrypt=self.args.encrypt, )
+                                              decrypt=self.args.decrypt, encrypt=self.args.encrypt, zfs_compressed=self.args.zfs_compressed )
             except Exception as e:
                 fail_count = fail_count + 1
                 source_dataset.error("FAILED: " + str(e))
@@ -464,7 +474,7 @@ class ZfsAutobackup:
                                   ssh_to=self.args.ssh_source, readonly=self.args.test,
                                   debug_output=self.args.debug_output, description=description, thinner=source_thinner)
             source_node.verbose(
-                "Selects all datasets that have property 'autobackup:{}=true' (or childs of datasets that have "
+                "Selects all datasets that have property 'autobackup:{}=true' (or children of datasets that have "
                 "'autobackup:{}=child')".format(
                     self.args.backup_name, self.args.backup_name))
 
@@ -500,7 +510,9 @@ class ZfsAutobackup:
             if not self.args.no_snapshot:
                 self.set_title("Snapshotting")
                 source_node.consistent_snapshot(source_datasets, source_node.new_snapshotname(),
-                                                min_changed_bytes=self.args.min_change)
+                                                min_changed_bytes=self.args.min_change,
+                                                pre_snapshot_cmds=self.args.pre_snapshot_cmd,
+                                                post_snapshot_cmds=self.args.post_snapshot_cmd)
 
             ################# sync
             # if target is specified, we sync the datasets, otherwise we just thin the source. (e.g. snapshot mode)

zfs_autobackup/ZfsDataset.py

@@ -503,7 +503,7 @@ class ZfsDataset:
 
         return self.from_names(names[1:])
 
-    def send_pipe(self, features, prev_snapshot, resume_token, show_progress, raw, send_properties, write_embedded, send_pipes):
+    def send_pipe(self, features, prev_snapshot, resume_token, show_progress, raw, send_properties, write_embedded, send_pipes, zfs_compressed):
         """returns a pipe with zfs send output for this snapshot
 
         resume_token: resume sending from this token. (in that case we don't
@@ -530,7 +530,7 @@ class ZfsDataset:
         if write_embedded and 'embedded_data' in features and "-e" in self.zfs_node.supported_send_options:
             cmd.append("--embed")  # WRITE_EMBEDDED, more compact stream
 
-        if "-c" in self.zfs_node.supported_send_options:
+        if zfs_compressed and "-c" in self.zfs_node.supported_send_options:
             cmd.append("--compressed")  # use compressed WRITE records
 
         # raw? (send over encrypted data in its original encrypted form without decrypting)
@@ -634,7 +634,7 @@ class ZfsDataset:
 
     def transfer_snapshot(self, target_snapshot, features, prev_snapshot, show_progress,
                           filter_properties, set_properties, ignore_recv_exit_code, resume_token,
-                          raw, send_properties, write_embedded, send_pipes, recv_pipes):
+                          raw, send_properties, write_embedded, send_pipes, recv_pipes, zfs_compressed):
         """transfer this snapshot to target_snapshot. specify prev_snapshot for
         incremental transfer
 
@@ -673,12 +673,13 @@ class ZfsDataset:
 
         # do it
         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, send_pipes=send_pipes)
+                              resume_token=resume_token, raw=raw, send_properties=send_properties, write_embedded=write_embedded, send_pipes=send_pipes, zfs_compressed=zfs_compressed)
         target_snapshot.recv_pipe(pipe, features=features, filter_properties=filter_properties,
                                   set_properties=set_properties, ignore_exit_code=ignore_recv_exit_code, recv_pipes=recv_pipes)
 
     def abort_resume(self):
         """abort current resume state"""
+        self.debug("Aborting resume")
         self.zfs_node.run(["zfs", "recv", "-A", self.name])
 
     def rollback(self):
@@ -873,9 +874,13 @@ class ZfsDataset:
             :type target_keeps: list of ZfsDataset
         """
 
-        # on source: destroy all obsoletes before common.
+        # on source: destroy all obsoletes before common. (since we cant send them anyways)
         # But after common, only delete snapshots that target also doesn't want
-        before_common = True
+        if common_snapshot:
+            before_common = True
+        else:
+            before_common = False
+
         for source_snapshot in self.snapshots:
             if common_snapshot and source_snapshot.snapshot_name == common_snapshot.snapshot_name:
                 before_common = False
@@ -887,8 +892,8 @@ class ZfsDataset:
 
         # on target: destroy everything thats obsolete, except common_snapshot
         for target_snapshot in target_dataset.snapshots:
-            if (target_snapshot in target_obsoletes) and (
-                    not common_snapshot or target_snapshot.snapshot_name != common_snapshot.snapshot_name):
+            if (target_snapshot in target_obsoletes) \
+                    and ( not common_snapshot or (target_snapshot.snapshot_name != common_snapshot.snapshot_name)):
                 if target_snapshot.exists:
                     target_snapshot.destroy()
 
@@ -901,14 +906,18 @@ class ZfsDataset:
         """
 
         if 'receive_resume_token' in target_dataset.properties:
-            resume_token = target_dataset.properties['receive_resume_token']
-            # not valid anymore?
-            resume_snapshot = self.get_resume_snapshot(resume_token)
-            if not resume_snapshot or start_snapshot.snapshot_name != resume_snapshot.snapshot_name:
-                target_dataset.verbose("Cant resume, resume token no longer valid.")
+            if start_snapshot==None:
+                target_dataset.verbose("Aborting resume, its obsolete.")
                 target_dataset.abort_resume()
             else:
-                return resume_token
+                resume_token = target_dataset.properties['receive_resume_token']
+                # not valid anymore
+                resume_snapshot = self.get_resume_snapshot(resume_token)
+                if not resume_snapshot or start_snapshot.snapshot_name != resume_snapshot.snapshot_name:
+                    target_dataset.verbose("Aborting resume, its no longer valid.")
+                    target_dataset.abort_resume()
+                else:
+                    return resume_token
 
     def _plan_sync(self, target_dataset, also_other_snapshots):
         """plan where to start syncing and what to sync and what to keep
@@ -963,7 +972,7 @@ class ZfsDataset:
 
     def sync_snapshots(self, target_dataset, features, show_progress, filter_properties, set_properties,
                        ignore_recv_exit_code, holds, rollback, decrypt, encrypt, also_other_snapshots,
-                       no_send, destroy_incompatible, send_pipes, recv_pipes):
+                       no_send, destroy_incompatible, send_pipes, recv_pipes, zfs_compressed):
         """sync this dataset's snapshots to target_dataset, while also thinning
         out old snapshots along the way.
 
@@ -1046,7 +1055,7 @@ class ZfsDataset:
                                                   filter_properties=active_filter_properties,
                                                   set_properties=active_set_properties,
                                                   ignore_recv_exit_code=ignore_recv_exit_code,
-                                                  resume_token=resume_token, write_embedded=write_embedded, raw=raw, send_properties=send_properties, send_pipes=send_pipes, recv_pipes=recv_pipes)
+                                                  resume_token=resume_token, write_embedded=write_embedded, raw=raw, send_properties=send_properties, send_pipes=send_pipes, recv_pipes=recv_pipes, zfs_compressed=zfs_compressed)
 
                 resume_token = None
 
@@ -1075,7 +1084,7 @@ class ZfsDataset:
                 source_snapshot.debug("skipped (target doesn't need it)")
                 # was it actually a resume?
                 if resume_token:
-                    target_dataset.debug("aborting resume, since we don't want that snapshot anymore")
+                    target_dataset.verbose("Aborting resume, we dont want that snapshot anymore.")
                     target_dataset.abort_resume()
                     resume_token = None
 

zfs_autobackup/ZfsNode.py

@@ -1,6 +1,7 @@
 # python 2 compatibility
 from __future__ import print_function
 import re
+import shlex
 import subprocess
 import sys
 import time
@@ -161,7 +162,7 @@ class ZfsNode(ExecuteNode):
         """determine uniq new snapshotname"""
         return self.backup_name + "-" + time.strftime("%Y%m%d%H%M%S")
 
-    def consistent_snapshot(self, datasets, snapshot_name, min_changed_bytes):
+    def consistent_snapshot(self, datasets, snapshot_name, min_changed_bytes, pre_snapshot_cmds=[], post_snapshot_cmds=[]):
         """create a consistent (atomic) snapshot of specified datasets, per pool.
         """
 
@@ -191,17 +192,32 @@ class ZfsNode(ExecuteNode):
             self.verbose("No changes anywhere: not creating snapshots.")
             return
 
-        # create consistent snapshot per pool
-        for (pool_name, snapshots) in pools.items():
-            cmd = ["zfs", "snapshot"]
+        try:
+            for cmd in pre_snapshot_cmds:
+                self.verbose("Running pre-snapshot-cmd")
+                self.run(cmd=shlex.split(cmd), readonly=False)
+
+            # create consistent snapshot per pool
+            for (pool_name, snapshots) in pools.items():
+                cmd = ["zfs", "snapshot"]
+
+                cmd.extend(map(lambda snapshot_: str(snapshot_), snapshots))
+
+                self.verbose("Creating snapshots {} in pool {}".format(snapshot_name, pool_name))
+                self.run(cmd, readonly=False)
+
+        finally:
+            for cmd in post_snapshot_cmds:
+                self.verbose("Running post-snapshot-cmd")
+                try:
+                    self.run(cmd=shlex.split(cmd), readonly=False)
+                except Exception as e:
+                    pass
 
-            cmd.extend(map(lambda snapshot_: str(snapshot_), snapshots))
 
-            self.verbose("Creating snapshots {} in pool {}".format(snapshot_name, pool_name))
-            self.run(cmd, readonly=False)
 
     def selected_datasets(self, exclude_received, exclude_paths):
-        """determine filesystems that should be backupped by looking at the special autobackup-property, systemwide
+        """determine filesystems that should be backed up by looking at the special autobackup-property, systemwide
 
            returns: list of ZfsDataset
         """