zfs_autobackup/bin/zfs-autobackup

#!/usr/bin/env python
# -*- coding: utf8 -*-

# (c)edwin@datux.nl  - Released under GPL V3
#
# Greetings from eth0 2019 :)

from __future__ import print_function

import os
import sys
import re
import subprocess
import time
import argparse
import select

colorama = False
if sys.stdout.isatty():
    try:
        import colorama
    except ImportError:
        colorama = False
        pass

VERSION = "3.0.1-beta4"
HEADER = "zfs-autobackup v{} - Copyright 2020 E.H.Eefting (edwin@datux.nl)".format(VERSION)


class Log:
    def __init__(self, show_debug=False, show_verbose=False):
        self.last_log = ""
        self.show_debug = show_debug
        self.show_verbose = show_verbose

    @staticmethod
    def error(txt):
        if colorama:
            print(colorama.Fore.RED + colorama.Style.BRIGHT + "! " + txt + colorama.Style.RESET_ALL, file=sys.stderr)
        else:
            print("! " + txt, file=sys.stderr)
        sys.stderr.flush()

    def verbose(self, txt):
        if self.show_verbose:
            if colorama:
                print(colorama.Style.NORMAL + "  " + txt + colorama.Style.RESET_ALL)
            else:
                print("  " + txt)
            sys.stdout.flush()

    def debug(self, txt):
        if self.show_debug:
            if colorama:
                print(colorama.Fore.GREEN + "# " + txt + colorama.Style.RESET_ALL)
            else:
                print("# " + txt)
            sys.stdout.flush()


class ThinnerRule:
    """a thinning schedule rule for Thinner"""

    TIME_NAMES = {
        'y': 3600 * 24 * 365.25,
        'm': 3600 * 24 * 30,
        'w': 3600 * 24 * 7,
        'd': 3600 * 24,
        'h': 3600,
        'min': 60,
        's': 1,
    }

    TIME_DESC = {
        'y': 'year',
        'm': 'month',
        'w': 'week',
        'd': 'day',
        'h': 'hour',
        'min': 'minute',
        's': 'second',
    }

    def __init__(self, rule_str):
        """parse scheduling string
            example:
                daily snapshot, remove after a week:     1d1w
                weekly snapshot, remove after a month:   1w1m
                monthly snapshot, remove after 6 months: 1m6m
                yearly snapshot, remove after 2 year:    1y2y
                keep all snapshots, remove after a day   1s1d
                keep nothing:                            1s1s

        """

        rule_str = rule_str.lower()
        matches = re.findall("([0-9]*)([a-z]*)([0-9]*)([a-z]*)", rule_str)[0]

        period_amount = int(matches[0])
        period_unit = matches[1]
        ttl_amount = int(matches[2])
        ttl_unit = matches[3]

        if period_unit not in self.TIME_NAMES:
            raise (Exception("Invalid period string in schedule: '{}'".format(rule_str)))

        if ttl_unit not in self.TIME_NAMES:
            raise (Exception("Invalid ttl string in schedule: '{}'".format(rule_str)))

        self.period = period_amount * self.TIME_NAMES[period_unit]
        self.ttl = ttl_amount * self.TIME_NAMES[ttl_unit]

        if self.period > self.ttl:
            raise (Exception("Period cant be longer than ttl in schedule: '{}'".format(rule_str)))

        self.rule_str = rule_str

        self.human_str = "Keep every {} {}{}, delete after {} {}{}.".format(
            period_amount, self.TIME_DESC[period_unit], period_amount != 1 and "s" or "", ttl_amount,
            self.TIME_DESC[ttl_unit], ttl_amount != 1 and "s" or "")

    def __str__(self):
        """get schedule as a schedule string"""

        return self.rule_str


class Thinner:
    """progressive thinner (universal, used for cleaning up snapshots)"""

    def __init__(self, schedule_str=""):
        """schedule_str: comma seperated list of ThinnerRules. A plain number specifies how many snapshots to always
        keep.
        """

        self.rules = []
        self.always_keep = 0

        if schedule_str == "":
            return

        rule_strs = schedule_str.split(",")
        for rule_str in rule_strs:
            if rule_str.isdigit():
                self.always_keep = int(rule_str)
                if self.always_keep < 0:
                    raise (Exception("Number of snapshots to keep cant be negative: {}".format(self.always_keep)))
            else:
                self.rules.append(ThinnerRule(rule_str))

    def human_rules(self):
        """get list of human readable rules"""
        ret = []
        if self.always_keep:
            ret.append("Keep the last {} snapshot{}.".format(self.always_keep, self.always_keep != 1 and "s" or ""))
        for rule in self.rules:
            ret.append(rule.human_str)

        return ret

    def thin(self, objects, keep_objects=None, now=None):
        """thin list of objects with current schedule rules. objects: list of objects to thin. every object should
        have timestamp attribute. keep_objects: objects to always keep (these should also be in normal objects list,
        so we can use them to perhaps delete other obsolete objects)

            return( keeps, removes )
        """

        if not keep_objects:
            keep_objects = []

        # always keep a number of the last objets?
        if self.always_keep:
            # all of them
            if len(objects) <= self.always_keep:
                return objects, []

            # determine which ones
            always_keep_objects = objects[-self.always_keep:]
        else:
            always_keep_objects = []

        # determine time blocks
        time_blocks = {}
        for rule in self.rules:
            time_blocks[rule.period] = {}

        if not now:
            now = int(time.time())

        keeps = []
        removes = []

        # traverse objects
        for thisobject in objects:
            # important they are ints!
            timestamp = int(thisobject.timestamp)
            age = int(now) - timestamp

            # store in the correct time blocks, per period-size, if not too old yet
            # e.g.: look if there is ANY timeblock that wants to keep this object
            keep = False
            for rule in self.rules:
                if age <= rule.ttl:
                    block_nr = int(timestamp / rule.period)
                    if block_nr not in time_blocks[rule.period]:
                        time_blocks[rule.period][block_nr] = True
                        keep = True

            # keep it according to schedule, or keep it because it is in the keep_objects list
            if keep or thisobject in keep_objects or thisobject in always_keep_objects:
                keeps.append(thisobject)
            else:
                removes.append(thisobject)

        return keeps, removes


# NOTE: this should inherit from (object) to function correctly with python 2.7
class CachedProperty(object):
    """ A property that is only computed once per instance and
    then stores the result in _cached_properties of the object.

        Source: https://github.com/bottlepy/bottle/commit/fa7733e075da0d790d809aa3d2f53071897e6f76
        """

    def __init__(self, func):
        self.__doc__ = getattr(func, '__doc__')
        self.func = func

    def __get__(self, obj, cls):
        if obj is None:
            return self

        propname = self.func.__name__

        if not hasattr(obj, '_cached_properties'):
            obj._cached_properties = {}

        if propname not in obj._cached_properties:
            obj._cached_properties[propname] = self.func(obj)
            # value = obj.__dict__[propname] = self.func(obj)

        return obj._cached_properties[propname]

    @staticmethod
    def clear(obj):
        """clears cache of obj"""
        if hasattr(obj, '_cached_properties'):
            obj._cached_properties = {}

    @staticmethod
    def is_cached(obj, propname):
        if hasattr(obj, '_cached_properties') and propname in obj._cached_properties:
            return True
        else:
            return False


class Logger:

    # simple logging stubs
    def debug(self, txt):
        print("DEBUG  : " + txt)

    def verbose(self, txt):
        print("VERBOSE: " + txt)

    def error(self, txt):
        print("ERROR  : " + txt)


class ExecuteNode(Logger):
    """an endpoint to execute local or remote commands via ssh"""

    def __init__(self, ssh_config=None, ssh_to=None, readonly=False, debug_output=False):
        """ssh_config: custom ssh config
           ssh_to: server you want to ssh to. none means local
           readonly: only execute commands that don't make any changes (useful for testing-runs)
           debug_output: show output and exit codes of commands in debugging output.
        """

        self.ssh_config = ssh_config
        self.ssh_to = ssh_to
        self.readonly = readonly
        self.debug_output = debug_output

    def __repr__(self):
        if self.ssh_to is None:
            return "(local)"
        else:
            return self.ssh_to

    def _parse_stdout(self, line):
        """parse stdout. can be overridden in subclass"""
        if self.debug_output:
            self.debug("STDOUT > " + line.rstrip())

    def _parse_stderr(self, line, hide_errors):
        """parse stderr. can be overridden in subclass"""
        if hide_errors:
            self.debug("STDERR > " + line.rstrip())
        else:
            self.error("STDERR > " + line.rstrip())

    def _parse_stderr_pipe(self, line, hide_errors):
        """parse stderr from pipe input process. can be overridden in subclass"""
        if hide_errors:
            self.debug("STDERR|> " + line.rstrip())
        else:
            self.error("STDERR|> " + line.rstrip())

    def run(self, cmd, inp=None, tab_split=False, valid_exitcodes=None, readonly=False, hide_errors=False, pipe=False,
            return_stderr=False):
        """run a command on the node cmd: the actual command, should be a list, where the first item is the command
        and the rest are parameters. input: Can be None, a string or a pipe-handle you got from another run()
        tab_split: split tabbed files in output into a list valid_exitcodes: list of valid exit codes for this
        command (checks exit code of both sides of a pipe) readonly: make this True if the command doesn't make any
        changes and is safe to execute in testmode hide_errors: don't show stderr output as error, instead show it as
        debugging output (use to hide expected errors) pipe: Instead of executing, return a pipe-handle to be used to
        input to another run() command. (just like a | in linux) return_stderr: return both stdout and stderr as a
        tuple. (only returns stderr from this side of the pipe)
        """

        if not valid_exitcodes:
            valid_exitcodes = [0]

        encoded_cmd = []

        # use ssh?
        if self.ssh_to is not None:
            encoded_cmd.append("ssh".encode('utf-8'))

            if self.ssh_config is not None:
                encoded_cmd.extend(["-F".encode('utf-8'), self.ssh_config.encode('utf-8')])

            encoded_cmd.append(self.ssh_to.encode('utf-8'))

            # 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)
            for arg in cmd:
                # add single quotes for remote commands to support spaces and other weird stuff (remote commands are
                # executed in a shell) and escape existing single quotes (bash needs ' to end the quoted string,
                # then a \' for the actual quote and then another ' to start a new quoted string) (and then python
                # needs the double \ to get a single \)
                encoded_cmd.append(("'" + arg.replace("'", "'\\''") + "'").encode('utf-8'))

        else:
            for arg in cmd:
                encoded_cmd.append(arg.encode('utf-8'))

        # debug and test stuff
        debug_txt = ""
        for c in encoded_cmd:
            debug_txt = debug_txt + " " + c.decode()

        if pipe:
            debug_txt = debug_txt + " |"

        if self.readonly and not readonly:
            self.debug("SKIP   > " + debug_txt)
        else:
            if pipe:
                self.debug("PIPE   > " + debug_txt)
            else:
                self.debug("RUN    > " + debug_txt)

        # determine stdin
        if inp is None:
            # NOTE: Not None, otherwise it reads stdin from terminal!
            stdin = subprocess.PIPE
        elif isinstance(inp, str) or type(inp) == 'unicode':
            self.debug("INPUT  > \n" + inp.rstrip())
            stdin = subprocess.PIPE
        elif isinstance(inp, subprocess.Popen):
            self.debug("Piping input")
            stdin = inp.stdout
        else:
            raise (Exception("Program error: Incompatible input"))

        if self.readonly and not readonly:
            # todo: what happens if input is piped?
            return

        # execute and parse/return results
        p = subprocess.Popen(encoded_cmd, env=os.environ, stdout=subprocess.PIPE, stdin=stdin, stderr=subprocess.PIPE)

        # Note: make streaming?
        if isinstance(inp, str) or type(inp) == 'unicode':
            p.stdin.write(inp.encode('utf-8'))

        if p.stdin:
            p.stdin.close()

        # return pipe
        if pipe:
            return p

        # handle all outputs
        if isinstance(inp, subprocess.Popen):
            selectors = [p.stdout, p.stderr, inp.stderr]
            inp.stdout.close()  # otherwise inputprocess wont exit when ours does
        else:
            selectors = [p.stdout, p.stderr]

        output_lines = []
        error_lines = []
        while True:
            (read_ready, write_ready, ex_ready) = select.select(selectors, [], [])
            eof_count = 0
            if p.stdout in read_ready:
                line = p.stdout.readline().decode('utf-8')
                if line != "":
                    if tab_split:
                        output_lines.append(line.rstrip().split('\t'))
                    else:
                        output_lines.append(line.rstrip())
                    self._parse_stdout(line)
                else:
                    eof_count = eof_count + 1
            if p.stderr in read_ready:
                line = p.stderr.readline().decode('utf-8')
                if line != "":
                    if tab_split:
                        error_lines.append(line.rstrip().split('\t'))
                    else:
                        error_lines.append(line.rstrip())
                    self._parse_stderr(line, hide_errors)
                else:
                    eof_count = eof_count + 1
            if isinstance(inp, subprocess.Popen) and (inp.stderr in read_ready):
                line = inp.stderr.readline().decode('utf-8')
                if line != "":
                    self._parse_stderr_pipe(line, hide_errors)
                else:
                    eof_count = eof_count + 1

            # stop if both processes are done and all filehandles are EOF:
            if (p.poll() is not None) and (
                    (not isinstance(inp, subprocess.Popen)) or inp.poll() is not None) and eof_count == len(selectors):
                break

        p.stderr.close()
        p.stdout.close()

        if self.debug_output:
            self.debug("EXIT   > {}".format(p.returncode))

        # handle piped process error output and exit codes
        if isinstance(inp, subprocess.Popen):
            inp.stderr.close()
            inp.stdout.close()

            if self.debug_output:
                self.debug("EXIT  |> {}".format(inp.returncode))
            if valid_exitcodes and inp.returncode not in valid_exitcodes:
                raise (subprocess.CalledProcessError(inp.returncode, "(pipe)"))

        if valid_exitcodes and p.returncode not in valid_exitcodes:
            raise (subprocess.CalledProcessError(p.returncode, encoded_cmd))

        if return_stderr:
            return output_lines, error_lines
        else:
            return output_lines


class ZfsPool:
    """a zfs pool"""

    def __init__(self, zfs_node, name):
        """name: name of the pool
        """

        self.zfs_node = zfs_node
        self.name = name

    def __repr__(self):
        return "{}: {}".format(self.zfs_node, self.name)

    def __str__(self):
        return self.name

    def __eq__(self, obj):
        if not isinstance(obj, ZfsPool):
            return False

        return self.name == obj.name

    def verbose(self, txt):
        self.zfs_node.verbose("zpool {}: {}".format(self.name, txt))

    def error(self, txt):
        self.zfs_node.error("zpool {}: {}".format(self.name, txt))

    def debug(self, txt):
        self.zfs_node.debug("zpool {}: {}".format(self.name, txt))

    @CachedProperty
    def properties(self):
        """all zpool properties"""

        self.debug("Getting zpool properties")

        cmd = [
            "zpool", "get", "-H", "-p", "all", self.name
        ]

        ret = {}

        for pair in self.zfs_node.run(tab_split=True, cmd=cmd, readonly=True, valid_exitcodes=[0]):
            if len(pair) == 4:
                ret[pair[1]] = pair[2]

        return ret

    @property
    def features(self):
        """get list of active zpool features"""

        ret = []
        for (key, value) in self.properties.items():
            if key.startswith("feature@"):
                feature = key.split("@")[1]
                if value == 'enabled' or value == 'active':
                    ret.append(feature)

        return ret


class ZfsDataset:
    """a zfs dataset (filesystem/volume/snapshot/clone)
    Note that a dataset doesn't have to actually exist (yet/anymore)
    Also most properties are cached for performance-reasons, but also to allow --test to function correctly.

    """

    # illegal properties per dataset type. these will be removed from --set-properties and --filter-properties
    ILLEGAL_PROPERTIES = {
        'filesystem': [],
        'volume': ["canmount"],
    }

    def __init__(self, zfs_node, name, force_exists=None):
        """name: full path of the zfs dataset exists: specify if you already know a dataset exists or not. for
        performance and testing reasons. (otherwise it will have to check with zfs list when needed)
        """
        self.zfs_node = zfs_node
        self.name = name  # full name
        self._virtual_snapshots = []
        self.invalidate()
        self.force_exists = force_exists

    def __repr__(self):
        return "{}: {}".format(self.zfs_node, self.name)

    def __str__(self):
        return self.name

    def __eq__(self, obj):
        if not isinstance(obj, ZfsDataset):
            return False

        return self.name == obj.name

    def verbose(self, txt):
        self.zfs_node.verbose("{}: {}".format(self.name, txt))

    def error(self, txt):
        self.zfs_node.error("{}: {}".format(self.name, txt))

    def debug(self, txt):
        self.zfs_node.debug("{}: {}".format(self.name, txt))

    def invalidate(self):
        """clear caches"""
        CachedProperty.clear(self)
        self.force_exists = None
        self._virtual_snapshots = []

    def split_path(self):
        """return the path elements as an array"""
        return self.name.split("/")

    def lstrip_path(self, count):
        """return name with first count components stripped"""
        return "/".join(self.split_path()[count:])

    def rstrip_path(self, count):
        """return name with last count components stripped"""
        return "/".join(self.split_path()[:-count])

    @property
    def filesystem_name(self):
        """filesystem part of the name (before the @)"""
        if self.is_snapshot:
            (filesystem, snapshot) = self.name.split("@")
            return filesystem
        else:
            return self.name

    @property
    def snapshot_name(self):
        """snapshot part of the name"""
        if not self.is_snapshot:
            raise (Exception("This is not a snapshot"))

        (filesystem, snapshot_name) = self.name.split("@")
        return snapshot_name

    @property
    def is_snapshot(self):
        """true if this dataset is a snapshot"""
        return self.name.find("@") != -1

    @CachedProperty
    def parent(self):
        """get zfs-parent of this dataset. for snapshots this means it will get the filesystem/volume that it belongs
        to. otherwise it will return the parent according to path

        we cache this so everything in the parent that is cached also stays.
        """
        if self.is_snapshot:
            return ZfsDataset(self.zfs_node, self.filesystem_name)
        else:
            return ZfsDataset(self.zfs_node, self.rstrip_path(1))

    def find_prev_snapshot(self, snapshot, other_snapshots=False):
        """find previous snapshot in this dataset. None if it doesn't exist.

        other_snapshots: set to true to also return snapshots that where not created by us. (is_ours)
        """

        if self.is_snapshot:
            raise (Exception("Please call this on a dataset."))

        index = self.find_snapshot_index(snapshot)
        while index:
            index = index - 1
            if other_snapshots or self.snapshots[index].is_ours():
                return self.snapshots[index]
        return None

    def find_next_snapshot(self, snapshot, other_snapshots=False):
        """find next snapshot in this dataset. None if it doesn't exist"""

        if self.is_snapshot:
            raise (Exception("Please call this on a dataset."))

        index = self.find_snapshot_index(snapshot)
        while index is not None and index < len(self.snapshots) - 1:
            index = index + 1
            if other_snapshots or self.snapshots[index].is_ours():
                return self.snapshots[index]
        return None

    @CachedProperty
    def exists(self):
        """check if dataset exists.
        Use force to force a specific value to be cached, if you already know. Useful for performance reasons"""

        if self.force_exists is not None:
            self.debug("Checking if filesystem exists: was forced to {}".format(self.force_exists))
            return self.force_exists
        else:
            self.debug("Checking if filesystem exists")

        return (self.zfs_node.run(tab_split=True, cmd=["zfs", "list", self.name], readonly=True, valid_exitcodes=[0, 1],
                                  hide_errors=True) and True)

    def create_filesystem(self, parents=False):
        """create a filesystem"""
        if parents:
            self.verbose("Creating filesystem and parents")
            self.zfs_node.run(["zfs", "create", "-p", self.name])
        else:
            self.verbose("Creating filesystem")
            self.zfs_node.run(["zfs", "create", self.name])

        self.force_exists = True

    def destroy(self, fail_exception=False):
        """destroy the dataset. by default failures are not an exception, so we can continue making backups"""

        self.verbose("Destroying")

        if self.is_snapshot:
            self.release()

        try:
            self.zfs_node.run(["zfs", "destroy", self.name])
            self.invalidate()
            self.force_exists = False
            return True
        except subprocess.CalledProcessError:
            if not fail_exception:
                return False
            else:
                raise

    @CachedProperty
    def properties(self):
        """all zfs properties"""

        cmd = [
            "zfs", "get", "-H", "-o", "property,value", "-p", "all", self.name
        ]

        if not self.exists:
            return {}

        self.debug("Getting zfs properties")

        ret = {}
        for pair in self.zfs_node.run(tab_split=True, cmd=cmd, readonly=True, valid_exitcodes=[0]):
            if len(pair) == 2:
                ret[pair[0]] = pair[1]

        return ret

    def is_changed(self, min_changed_bytes=1):
        """dataset is changed since ANY latest snapshot ?"""
        self.debug("Checking if dataset is changed")

        if min_changed_bytes == 0:
            return True

        if int(self.properties['written']) < min_changed_bytes:
            return False
        else:
            return True

    def is_ours(self):
        """return true if this snapshot is created by this backup_name"""
        if re.match("^" + self.zfs_node.backup_name + "-[0-9]*$", self.snapshot_name):
            return True
        else:
            return False

    @property
    def _hold_name(self):
        return "zfs_autobackup:" + self.zfs_node.backup_name

    @property
    def holds(self):
        """get list of holds for dataset"""

        output = self.zfs_node.run(["zfs", "holds", "-H", self.name], valid_exitcodes=[0], tab_split=True,
                                   readonly=True)
        return map(lambda fields: fields[1], output)

    def is_hold(self):
        """did we hold this snapshot?"""
        return self._hold_name in self.holds

    def hold(self):
        """hold dataset"""
        self.debug("holding")
        self.zfs_node.run(["zfs", "hold", self._hold_name, self.name], valid_exitcodes=[0, 1])

    def release(self):
        """release dataset"""
        if self.zfs_node.readonly or self.is_hold():
            self.debug("releasing")
            self.zfs_node.run(["zfs", "release", self._hold_name, self.name], valid_exitcodes=[0, 1])

    @property
    def timestamp(self):
        """get timestamp from snapshot name. Only works for our own snapshots with the correct format."""
        time_str = re.findall("^.*-([0-9]*)$", self.snapshot_name)[0]
        if len(time_str) != 14:
            raise (Exception("Snapshot has invalid timestamp in name: {}".format(self.snapshot_name)))

        # new format:
        time_secs = time.mktime(time.strptime(time_str, "%Y%m%d%H%M%S"))
        return time_secs

    def from_names(self, names):
        """convert a list of names to a list ZfsDatasets for this zfs_node"""
        ret = []
        for name in names:
            ret.append(ZfsDataset(self.zfs_node, name))

        return ret

    # def add_virtual_snapshot(self, snapshot):
    #     """pretend a snapshot exists (usefull in test mode)"""
    #
    #     # NOTE: we could just call self.snapshots.append() but this would trigger a zfs list which is not always needed.
    #     if CachedProperty.is_cached(self, 'snapshots'):
    #         # already cached so add it
    #         print ("ADDED")
    #         self.snapshots.append(snapshot)
    #     else:
    #         # self.snapshots will add it when requested
    #         print ("ADDED VIRT")
    #         self._virtual_snapshots.append(snapshot)

    @CachedProperty
    def snapshots(self):
        """get all snapshots of this dataset"""


        if not self.exists:
            return []

        self.debug("Getting snapshots")

        cmd = [
            "zfs", "list", "-d", "1", "-r", "-t", "snapshot", "-H", "-o", "name", self.name
        ]

        return self.from_names(self.zfs_node.run(cmd=cmd, readonly=True))

    @property
    def our_snapshots(self):
        """get list of snapshots creates by us of this dataset"""
        ret = []
        for snapshot in self.snapshots:
            if snapshot.is_ours():
                ret.append(snapshot)

        return ret

    def find_snapshot(self, snapshot):
        """find snapshot by snapshot (can be a snapshot_name or a different ZfsDataset )"""

        if not isinstance(snapshot, ZfsDataset):
            snapshot_name = snapshot
        else:
            snapshot_name = snapshot.snapshot_name

        for snapshot in self.snapshots:
            if snapshot.snapshot_name == snapshot_name:
                return snapshot

        return None

    def find_snapshot_index(self, snapshot):
        """find snapshot index by snapshot (can be a snapshot_name or ZfsDataset)"""

        if not isinstance(snapshot, ZfsDataset):
            snapshot_name = snapshot
        else:
            snapshot_name = snapshot.snapshot_name

        index = 0
        for snapshot in self.snapshots:
            if snapshot.snapshot_name == snapshot_name:
                return index
            index = index + 1

        return None

    @CachedProperty
    def written_since_ours(self):
        """get number of bytes written since our last snapshot"""

        latest_snapshot = self.our_snapshots[-1]

        self.debug("Getting bytes written since our last snapshot")
        cmd = ["zfs", "get", "-H", "-ovalue", "-p", "written@" + str(latest_snapshot), self.name]

        output = self.zfs_node.run(readonly=True, tab_split=False, cmd=cmd, valid_exitcodes=[0])

        return int(output[0])

    def is_changed_ours(self, min_changed_bytes=1):
        """dataset is changed since OUR latest snapshot?"""

        if min_changed_bytes == 0:
            return True

        if not self.our_snapshots:
            return True

        # NOTE: filesystems can have a very small amount written without actual changes in some cases
        if self.written_since_ours < min_changed_bytes:
            return False

        return True

    @CachedProperty
    def recursive_datasets(self, types="filesystem,volume"):
        """get all (non-snapshot) datasets recursively under us"""

        self.debug("Getting all recursive datasets under us")

        names = self.zfs_node.run(tab_split=False, readonly=True, valid_exitcodes=[0], cmd=[
            "zfs", "list", "-r", "-t", types, "-o", "name", "-H", self.name
        ])

        return self.from_names(names[1:])

    @CachedProperty
    def datasets(self, types="filesystem,volume"):
        """get all (non-snapshot) datasets directly under us"""

        self.debug("Getting all datasets under us")

        names = self.zfs_node.run(tab_split=False, readonly=True, valid_exitcodes=[0], cmd=[
            "zfs", "list", "-r", "-t", types, "-o", "name", "-H", "-d", "1", self.name
        ])

        return self.from_names(names[1:])

    def send_pipe(self, features, prev_snapshot=None, resume_token=None, show_progress=False, raw=False):
        """returns a pipe with zfs send output for this snapshot

        resume_token: resume sending from this token. (in that case we don't need to know snapshot names)

        """
        # build source command
        cmd = []

        cmd.extend(["zfs", "send", ])

        # all kind of performance options:
        if 'large_blocks' in features and "-L" in self.zfs_node.supported_send_options:
            cmd.append("-L")  # large block support (only if recordsize>128k which is seldomly used)

        if 'embedded_data' in features and "-e" in self.zfs_node.supported_send_options:
            cmd.append("-e")  # WRITE_EMBEDDED, more compact stream

        if "-c" in self.zfs_node.supported_send_options:
            cmd.append("-c")  # use compressed WRITE records

        # NOTE: performance is usually worse with this option, according to manual
        # also -D will be depricated in newer ZFS versions
        # if not resume:
        #     if "-D" in self.zfs_node.supported_send_options:
        #         cmd.append("-D") # dedupped stream, sends less duplicate data

        # raw? (for encryption)
        if raw:
            cmd.append("--raw")

        # progress output
        if show_progress:
            cmd.append("-v")
            cmd.append("-P")

        # resume a previous send? (don't need more parameters in that case)
        if resume_token:
            cmd.extend(["-t", resume_token])

        else:
            # send properties
            cmd.append("-p")

            # incremental?
            if prev_snapshot:
                cmd.extend(["-i", "@" + prev_snapshot.snapshot_name])

            cmd.append(self.name)

        # if args.buffer and args.ssh_source!="local":
        #     cmd.append("|mbuffer -m {}".format(args.buffer))

        # NOTE: this doesn't start the send yet, it only returns a subprocess.Pipe
        return self.zfs_node.run(cmd, pipe=True)

    def recv_pipe(self, pipe, features, 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 resulting zfs command is the same, only our object cache is invalidated differently.
        """

        if set_properties is None:
            set_properties = []

        if filter_properties is None:
            filter_properties = []

        # build target command
        cmd = []

        cmd.extend(["zfs", "recv"])

        # don't mount filesystem that is received
        cmd.append("-u")

        for property_ in filter_properties:
            cmd.extend(["-x", property_])

        for property_ in set_properties:
            cmd.extend(["-o", property_])

        # verbose output
        cmd.append("-v")

        if 'extensible_dataset' in features and "-s" in self.zfs_node.supported_recv_options:
            # support resuming
            self.debug("Enabled resume support")
            cmd.append("-s")

        cmd.append(self.filesystem_name)

        if ignore_exit_code:
            valid_exitcodes = []
        else:
            valid_exitcodes = [0]

        self.zfs_node.reset_progress()
        self.zfs_node.run(cmd, inp=pipe, valid_exitcodes=valid_exitcodes)

        # invalidate cache, but we at least know we exist now
        self.invalidate()

        # in test mode we assume everything was ok and it exists
        if self.zfs_node.readonly:
            self.force_exists = True

        # check if transfer was really ok (exit codes have been wrong before due to bugs in zfs-utils and can be
        # ignored by some parameters)
        if not self.exists:
            self.error("error during transfer")
            raise (Exception("Target doesn't exist after transfer, something went wrong."))

        # if args.buffer and  args.ssh_target!="local":
        #     cmd.append("|mbuffer -m {}".format(args.buffer))

    def transfer_snapshot(self, target_snapshot, features, prev_snapshot=None, show_progress=False,
                          filter_properties=None, set_properties=None, ignore_recv_exit_code=False, resume_token=None,
                          raw=False):
        """transfer this snapshot to target_snapshot. specify prev_snapshot for incremental transfer

        connects a send_pipe() to recv_pipe()
        """

        if set_properties is None:
            set_properties = []
        if filter_properties is None:
            filter_properties = []

        self.debug("Transfer snapshot to {}".format(target_snapshot.filesystem_name))

        if resume_token:
            target_snapshot.verbose("resuming")

        # initial or increment
        if not prev_snapshot:
            target_snapshot.verbose("receiving full".format(self.snapshot_name))
        else:
            # incremental
            target_snapshot.verbose("receiving incremental".format(self.snapshot_name))

        # do it
        pipe = self.send_pipe(features=features, show_progress=show_progress, prev_snapshot=prev_snapshot,
                              resume_token=resume_token, raw=raw)
        target_snapshot.recv_pipe(pipe, features=features, filter_properties=filter_properties,
                                  set_properties=set_properties, ignore_exit_code=ignore_recv_exit_code)

    def abort_resume(self):
        """abort current resume state"""
        self.zfs_node.run(["zfs", "recv", "-A", self.name])

    def rollback(self):
        """rollback to latest existing snapshot on this dataset"""

        for snapshot in reversed(self.snapshots):
            if snapshot.exists:
                self.debug("Rolling back")
                self.zfs_node.run(["zfs", "rollback", snapshot.name])
                return

    def get_resume_snapshot(self, resume_token):
        """returns snapshot that will be resumed by this resume token (run this on source with target-token)"""
        # use zfs send -n option to determine this
        # NOTE: on smartos stderr, on linux stdout
        (stdout, stderr) = self.zfs_node.run(["zfs", "send", "-t", resume_token, "-n", "-v"], valid_exitcodes=[0, 255],
                                             readonly=True, return_stderr=True)
        if stdout:
            lines = stdout
        else:
            lines = stderr
        for line in lines:
            matches = re.findall("toname = .*@(.*)", line)
            if matches:
                snapshot_name = matches[0]
                snapshot = ZfsDataset(self.zfs_node, self.filesystem_name + "@" + snapshot_name)
                snapshot.debug("resume token belongs to this snapshot")
                return snapshot

        return None

    def thin_list(self, keeps=None, ignores=None):
        """determines list of snapshots that should be kept or deleted based on the thinning schedule. cull the herd!
        keep: list of snapshots to always keep (usually the last) ignores: snapshots to completely ignore (usually
        incompatible target snapshots that are going to be destroyed anyway)

        returns: ( keeps, obsoletes )
        """

        if ignores is None:
            ignores = []
        if keeps is None:
            keeps = []

        snapshots = [snapshot for snapshot in self.our_snapshots if snapshot not in ignores]

        return self.zfs_node.thinner.thin(snapshots, keep_objects=keeps)

    def thin(self, skip_holds=False):
        """destroys snapshots according to thin_list, except last snapshot"""

        (keeps, obsoletes) = self.thin_list(keeps=self.our_snapshots[-1:])
        for obsolete in obsoletes:
            if skip_holds and obsolete.is_hold():
                obsolete.verbose("Keeping (common snapshot)")
            else:
                obsolete.destroy()
                self.snapshots.remove(obsolete)

    def find_common_snapshot(self, target_dataset):
        """find latest common snapshot between us and target
        returns None if its an initial transfer
        """
        if not target_dataset.snapshots:
            # target has nothing yet
            return None
        else:
            # snapshot=self.find_snapshot(target_dataset.snapshots[-1].snapshot_name)

            # if not snapshot:
            # try to common snapshot
            for source_snapshot in reversed(self.snapshots):
                if target_dataset.find_snapshot(source_snapshot):
                    source_snapshot.debug("common snapshot")
                    return source_snapshot
            target_dataset.error("Cant find common snapshot with source.")
            raise (Exception("You probably need to delete the target dataset to fix this."))

    def find_start_snapshot(self, common_snapshot, other_snapshots):
        """finds first snapshot to send"""

        if not common_snapshot:
            if not self.snapshots:
                start_snapshot = None
            else:
                # start from beginning
                start_snapshot = self.snapshots[0]

                if not start_snapshot.is_ours() and not other_snapshots:
                    # try to start at a snapshot thats ours
                    start_snapshot = self.find_next_snapshot(start_snapshot, other_snapshots)
        else:
            start_snapshot = self.find_next_snapshot(common_snapshot, other_snapshots)

        return start_snapshot

    def find_incompatible_snapshots(self, common_snapshot):
        """returns a list of snapshots that is incompatible for a zfs recv onto the common_snapshot.
        all direct followup snapshots with written=0 are compatible."""

        ret = []

        if common_snapshot and self.snapshots:
            followup = True
            for snapshot in self.snapshots[self.find_snapshot_index(common_snapshot) + 1:]:
                if not followup or int(snapshot.properties['written']) != 0:
                    followup = False
                    ret.append(snapshot)

        return ret

    def get_allowed_properties(self, filter_properties, set_properties):
        """only returns lists of allowed properties for this dataset type"""

        allowed_filter_properties = []
        allowed_set_properties = []
        illegal_properties = self.ILLEGAL_PROPERTIES[self.properties['type']]
        for set_property in set_properties:
            (property_, value) = set_property.split("=")
            if property_ not in illegal_properties:
                allowed_set_properties.append(set_property)

        for filter_property in filter_properties:
            if filter_property not in illegal_properties:
                allowed_filter_properties.append(filter_property)

        return allowed_filter_properties, allowed_set_properties

    def sync_snapshots(self, target_dataset, features, show_progress=False, filter_properties=None, set_properties=None,
                       ignore_recv_exit_code=False, holds=True, rollback=False, raw=False, other_snapshots=False,
                       no_send=False, destroy_incompatible=False):
        """sync this dataset's snapshots to target_dataset, while also thinning out old snapshots along the way."""

        if set_properties is None:
            set_properties = []
        if filter_properties is None:
            filter_properties = []

        # determine common and start snapshot
        target_dataset.debug("Determining start snapshot")
        common_snapshot = self.find_common_snapshot(target_dataset)
        start_snapshot = self.find_start_snapshot(common_snapshot, other_snapshots)
        # should be destroyed before attempting zfs recv:
        incompatible_target_snapshots = target_dataset.find_incompatible_snapshots(common_snapshot)

        # make target snapshot list the same as source, by adding virtual non-existing ones to the list.
        target_dataset.debug("Creating virtual target snapshots")
        source_snapshot = start_snapshot
        while source_snapshot:
            # create virtual target snapshot
            virtual_snapshot = ZfsDataset(target_dataset.zfs_node,
                                          target_dataset.filesystem_name + "@" + source_snapshot.snapshot_name,
                                          force_exists=False)
            target_dataset.snapshots.append(virtual_snapshot)
            source_snapshot = self.find_next_snapshot(source_snapshot, other_snapshots)

        # now let thinner decide what we want on both sides as final state (after all transfers are done)
        if self.our_snapshots:
            self.debug("Create thinning list")
            (source_keeps, source_obsoletes) = self.thin_list(keeps=[self.our_snapshots[-1]])
        else:
            source_obsoletes = []

        if target_dataset.our_snapshots:
            (target_keeps, target_obsoletes) = target_dataset.thin_list(keeps=[target_dataset.our_snapshots[-1]],
                                                                        ignores=incompatible_target_snapshots)
        else:
            target_keeps = []
            target_obsoletes = []

        # on source: destroy all obsoletes before common. but after common, only delete snapshots that target also
        # doesn't want to explicitly keep
        before_common = True
        for source_snapshot in self.snapshots:
            if common_snapshot and source_snapshot.snapshot_name == common_snapshot.snapshot_name:
                before_common = False
                # never destroy common snapshot
            else:
                target_snapshot = target_dataset.find_snapshot(source_snapshot)
                if (source_snapshot in source_obsoletes) and (before_common or (target_snapshot not in target_keeps)):
                    source_snapshot.destroy()

        # 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.exists:
                    target_snapshot.destroy()

        # now actually transfer the snapshots, if we want
        if no_send:
            return

        # resume?
        resume_token = None
        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.")
                target_dataset.abort_resume()
                resume_token = None

        # incompatible target snapshots?
        if incompatible_target_snapshots:
            if not destroy_incompatible:
                for snapshot in incompatible_target_snapshots:
                    snapshot.error("Incompatible snapshot")
                raise (Exception("Please destroy incompatible snapshots or use --destroy-incompatible."))
            else:
                for snapshot in incompatible_target_snapshots:
                    snapshot.verbose("Incompatible snapshot")
                    snapshot.destroy()
                    target_dataset.snapshots.remove(snapshot)

        # rollback target to latest?
        if rollback:
            target_dataset.rollback()

        # now actually transfer the snapshots
        prev_source_snapshot = common_snapshot
        source_snapshot = start_snapshot
        while source_snapshot:
            target_snapshot = target_dataset.find_snapshot(source_snapshot)  # still virtual

            # does target actually want it?
            if target_snapshot not in target_obsoletes:
                # NOTE: should we let transfer_snapshot handle this?
                (allowed_filter_properties, allowed_set_properties) = self.get_allowed_properties(filter_properties,
                                                                                                  set_properties)
                source_snapshot.transfer_snapshot(target_snapshot, features=features,
                                                  prev_snapshot=prev_source_snapshot, show_progress=show_progress,
                                                  filter_properties=allowed_filter_properties,
                                                  set_properties=allowed_set_properties,
                                                  ignore_recv_exit_code=ignore_recv_exit_code,
                                                  resume_token=resume_token, raw=raw)
                resume_token = None

                # hold the new common snapshots and release the previous ones
                if holds:
                    target_snapshot.hold()
                    source_snapshot.hold()

                if prev_source_snapshot:
                    if holds:
                        prev_source_snapshot.release()
                        target_dataset.find_snapshot(prev_source_snapshot).release()

                # we may now destroy the previous source snapshot if its obsolete
                if prev_source_snapshot in source_obsoletes:
                    prev_source_snapshot.destroy()

                    # destroy the previous target snapshot if obsolete (usually this is only the common_snapshot,
                    # the rest was already destroyed or will not be send)
                prev_target_snapshot = target_dataset.find_snapshot(prev_source_snapshot)
                if prev_target_snapshot in target_obsoletes:
                    prev_target_snapshot.destroy()

                prev_source_snapshot = source_snapshot
            else:
                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.abort_resume()
                    resume_token = None

            source_snapshot = self.find_next_snapshot(source_snapshot, other_snapshots)


class ZfsNode(ExecuteNode):
    """a node that contains zfs datasets. implements global (systemwide/pool wide) zfs commands"""

    def __init__(self, backup_name, logger, ssh_config=None, ssh_to=None, readonly=False, description="",
                 debug_output=False, thinner=Thinner()):
        self.backup_name = backup_name
        if not description and ssh_to:
            self.description = ssh_to
        else:
            self.description = description

        self.logger = logger

        if ssh_config:
            self.verbose("Using custom SSH config: {}".format(ssh_config))

        if ssh_to:
            self.verbose("Datasets on: {}".format(ssh_to))
        else:
            self.verbose("Datasets are local")

        rules = thinner.human_rules()
        if rules:
            for rule in rules:
                self.verbose(rule)
        else:
            self.verbose("Keep no old snaphots")

        self.thinner = thinner

        # list of ZfsPools
        self.__pools = {}

        self._progress_total_bytes = 0
        self._progress_start_time = time.time()

        ExecuteNode.__init__(self, ssh_config=ssh_config, ssh_to=ssh_to, readonly=readonly, debug_output=debug_output)

    @CachedProperty
    def supported_send_options(self):
        """list of supported options, for optimizing sends"""
        # not every zfs implementation supports them all

        ret = []
        for option in ["-L", "-e", "-c"]:
            if self.valid_command(["zfs", "send", option, "zfs_autobackup_option_test"]):
                ret.append(option)
        return ret

    @CachedProperty
    def supported_recv_options(self):
        """list of supported options"""
        # not every zfs implementation supports them all

        ret = []
        for option in ["-s"]:
            if self.valid_command(["zfs", "recv", option, "zfs_autobackup_option_test"]):
                ret.append(option)
        return ret

    def valid_command(self, cmd):
        """test if a specified zfs options are valid exit code. use this to determine support options"""

        try:
            self.run(cmd, hide_errors=True, valid_exitcodes=[0, 1])
        except subprocess.CalledProcessError:
            return False

        return True

    # TODO: also create a get_zfs_dataset() function that stores all the objects in a dict. This should optimize
    #  caching a bit and is more consistent.
    def get_zfs_pool(self, name):
        """get a ZfsPool() object from specified name. stores objects internally to enable caching"""

        return self.__pools.setdefault(name, ZfsPool(self, name))

    def reset_progress(self):
        """reset progress output counters"""
        self._progress_total_bytes = 0
        self._progress_start_time = time.time()

    def parse_zfs_progress(self, line, hide_errors, prefix):
        """try to parse progress output of zfs recv -Pv, and don't show it as error to the user """

        # is it progress output?
        progress_fields = line.rstrip().split("\t")

        if (line.find("nvlist version") == 0 or
                line.find("resume token contents") == 0 or
                len(progress_fields) != 1 or
                line.find("skipping ") == 0 or
                re.match("send from .*estimated size is ", line)):

            # always output for debugging offcourse
            self.debug(prefix + line.rstrip())

            # actual useful info
            if len(progress_fields) >= 3:
                if progress_fields[0] == 'full' or progress_fields[0] == 'size':
                    self._progress_total_bytes = int(progress_fields[2])
                elif progress_fields[0] == 'incremental':
                    self._progress_total_bytes = int(progress_fields[3])
                else:
                    bytes_ = int(progress_fields[1])
                    if self._progress_total_bytes:
                        percentage = min(100, int(bytes_ * 100 / self._progress_total_bytes))
                        speed = int(bytes_ / (time.time() - self._progress_start_time) / (1024 * 1024))
                        bytes_left = self._progress_total_bytes - bytes_
                        minutes_left = int((bytes_left / (bytes_ / (time.time() - self._progress_start_time))) / 60)

                        print(">>> {}% {}MB/s (total {}MB, {} minutes left)     \r".format(percentage, speed, int(
                            self._progress_total_bytes / (1024 * 1024)), minutes_left), end='', file=sys.stderr)
                        sys.stderr.flush()

            return

        # still do the normal stderr output handling
        if hide_errors:
            self.debug(prefix + line.rstrip())
        else:
            self.error(prefix + line.rstrip())

    def _parse_stderr_pipe(self, line, hide_errors):
        self.parse_zfs_progress(line, hide_errors, "STDERR|> ")

    def _parse_stderr(self, line, hide_errors):
        self.parse_zfs_progress(line, hide_errors, "STDERR > ")

    def verbose(self, txt):
        self.logger.verbose("{} {}".format(self.description, txt))

    def error(self, txt):
        self.logger.error("{} {}".format(self.description, txt))

    def debug(self, txt):
        self.logger.debug("{} {}".format(self.description, txt))

    def new_snapshotname(self):
        """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):
        """create a consistent (atomic) snapshot of specified datasets, per pool.
        """

        pools = {}

        # collect snapshots that we want to make, per pool
        # self.debug(datasets)
        for dataset in datasets:
            if not dataset.is_changed_ours(min_changed_bytes):
                dataset.verbose("No changes since {}".format(dataset.our_snapshots[-1].snapshot_name))
                continue

            # force_exist, since we're making it
            snapshot = ZfsDataset(dataset.zfs_node, dataset.name + "@" + snapshot_name, force_exists=True)

            pool = dataset.split_path()[0]
            if pool not in pools:
                pools[pool] = []

            pools[pool].append(snapshot)

            # update cache, but try to prevent an unneeded zfs list
            if self.readonly or CachedProperty.is_cached(dataset, 'snapshots'):
                dataset.snapshots.append(snapshot)  # NOTE: this will trigger zfs list if its not cached

        if not pools:
            self.verbose("No changes anywhere: not creating snapshots.")
            return

        # 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)

    @CachedProperty
    def selected_datasets(self):
        """determine filesystems that should be backupped by looking at the special autobackup-property, systemwide

           returns: list of ZfsDataset
        """

        self.debug("Getting selected datasets")

        # get all source filesystems that have the backup property
        lines = self.run(tab_split=True, readonly=True, cmd=[
            "zfs", "get", "-t", "volume,filesystem", "-o", "name,value,source", "-s", "local,inherited", "-H",
            "autobackup:" + self.backup_name
        ])

        # determine filesystems that should be actually backupped
        selected_filesystems = []
        direct_filesystems = []
        for line in lines:
            (name, value, source) = line
            dataset = ZfsDataset(self, name)

            if value == "false":
                dataset.verbose("Ignored (disabled)")

            else:
                if source == "local" and (value == "true" or value == "child"):
                    direct_filesystems.append(name)

                if source == "local" and value == "true":
                    dataset.verbose("Selected (direct selection)")
                    selected_filesystems.append(dataset)
                elif source.find("inherited from ") == 0 and (value == "true" or value == "child"):
                    inherited_from = re.sub("^inherited from ", "", source)
                    if inherited_from in direct_filesystems:
                        selected_filesystems.append(dataset)
                        dataset.verbose("Selected (inherited selection)")
                    else:
                        dataset.debug("Ignored (already a backup)")
                else:
                    dataset.verbose("Ignored (only childs)")

        return selected_filesystems


class ZfsAutobackup:
    """main class"""

    def __init__(self, argv, print_arguments=True):

        # helps with investigating failed regression tests:
        if print_arguments:
            print("ARGUMENTS: " + " ".join(argv))

        parser = argparse.ArgumentParser(
            description=HEADER,
            epilog='When a filesystem fails, zfs_backup will continue and report the number of failures at that end. '
                   'Also the exit code will indicate the number of failures.')
        parser.add_argument('--ssh-config', default=None, help='Custom ssh client config')
        parser.add_argument('--ssh-source', default=None,
                            help='Source host to get backup from. (user@hostname) Default %(default)s.')
        parser.add_argument('--ssh-target', default=None,
                            help='Target host to push backup to. (user@hostname) Default  %(default)s.')
        parser.add_argument('--keep-source', type=str, default="10,1d1w,1w1m,1m1y",
                            help='Thinning schedule for old source snapshots. Default: %(default)s')
        parser.add_argument('--keep-target', type=str, default="10,1d1w,1w1m,1m1y",
                            help='Thinning schedule for old target snapshots. Default: %(default)s')

        parser.add_argument('backup_name', metavar='backup-name',
                            help='Name of the backup (you should set the zfs property "autobackup:backup-name" to '
                                 'true on filesystems you want to backup')
        parser.add_argument('target_path', metavar='target-path', default=None, nargs='?',
                            help='Target ZFS filesystem (optional: if not specified, zfs-autobackup will only operate '
                                 'as snapshot-tool on source)')

        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',
                            help='Don\'t create new snapshots (useful for finishing uncompleted backups, or cleanups)')
        parser.add_argument('--no-send', action='store_true',
                            help='Don\'t send snapshots (useful for cleanups, or if you want a serperate send-cronjob)')
        #        parser.add_argument('--no-thinning', action='store_true', help='Don\'t run the thinner.')
        parser.add_argument('--min-change', type=int, default=1,
                            help='Number of bytes written after which we consider a dataset changed (default %('
                                 'default)s)')
        parser.add_argument('--allow-empty', action='store_true',
                            help='If nothing has changed, still create empty snapshots. (same as --min-change=0)')
        parser.add_argument('--ignore-replicated', action='store_true',
                            help='Ignore datasets that seem to be replicated some other way. (No changes since '
                                 'lastest snapshot. Useful for proxmox HA replication)')
        parser.add_argument('--no-holds', action='store_true',
                            help='Don\'t hold snapshots. (Faster)')

        parser.add_argument('--resume', action='store_true', help=argparse.SUPPRESS)
        parser.add_argument('--strip-path', default=0, type=int,
                            help='Number of directories to strip from target path (use 1 when cloning zones between 2 '
                                 'SmartOS machines)')
        # parser.add_argument('--buffer', default="",  help='Use mbuffer with specified size to speedup zfs transfer.
        # (e.g. --buffer 1G) Will also show nice progress output.')

        parser.add_argument('--clear-refreservation', action='store_true',
                            help='Filter "refreservation" property. (recommended, safes space. same as '
                                 '--filter-properties refreservation)')
        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', 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', type=str,
                            help='List of propererties to override when receiving filesystems. (you can still restore '
                                 'them with zfs inherit -S)')
        parser.add_argument('--rollback', action='store_true',
                            help='Rollback changes to the latest target snapshot before starting. (normally you can '
                                 'prevent changes by setting the readonly property on the target_path to on)')
        parser.add_argument('--destroy-incompatible', action='store_true',
                            help='Destroy incompatible snapshots on target. Use with care! (implies --rollback)')
        parser.add_argument('--destroy-missing', type=str, default=None,
                            help='Destroy datasets on target that are missing on the source. Specify the time since '
                                 'the last snapshot, e.g: --destroy-missing 30d')
        parser.add_argument('--ignore-transfer-errors', action='store_true',
                            help='Ignore transfer errors (still checks if received filesystem exists. useful for '
                                 'acltype errors)')
        parser.add_argument('--raw', action='store_true',
                            help='For encrypted datasets, send data exactly as it exists on disk.')

        parser.add_argument('--test', action='store_true',
                            help='dont change anything, just show what would be done (still does all read-only '
                                 'operations)')
        parser.add_argument('--verbose', action='store_true', help='verbose output')
        parser.add_argument('--debug', action='store_true',
                            help='Show zfs commands that are executed, stops after an exception.')
        parser.add_argument('--debug-output', action='store_true',
                            help='Show zfs commands and their output/exit codes. (noisy)')
        parser.add_argument('--progress', action='store_true',
                            help='show zfs progress output (to stderr). Enabled by default on ttys.')

        # note args is the only global variable we use, since its a global readonly setting anyway
        args = parser.parse_args(argv)

        self.args = args

        if sys.stderr.isatty():
            args.progress = True

        if args.debug_output:
            args.debug = True

        if self.args.test:
            self.args.verbose = True

        if args.allow_empty:
            args.min_change = 0

        if args.destroy_incompatible:
            args.rollback = True

        self.log = Log(show_debug=self.args.debug, show_verbose=self.args.verbose)

        if args.resume:
            self.verbose("NOTE: The --resume option isn't needed anymore (its autodetected now)")

        if args.target_path is not None and args.target_path[0] == "/":
            self.log.error("Target should not start with a /")
            sys.exit(255)

    def verbose(self, txt):
        self.log.verbose(txt)

    def error(self, txt):
        self.log.error(txt)

    def debug(self, txt):
        self.log.debug(txt)

    def set_title(self, title):
        self.log.verbose("")
        self.log.verbose("#### " + title)

    # sync datasets, or thin-only on both sides
    # target is needed for this.
    def sync_datasets(self, source_node, source_datasets):

        description = "[Target]"

        self.set_title("Target settings")

        target_thinner = Thinner(self.args.keep_target)
        target_node = ZfsNode(self.args.backup_name, self, ssh_config=self.args.ssh_config, ssh_to=self.args.ssh_target,
                              readonly=self.args.test, debug_output=self.args.debug_output, description=description,
                              thinner=target_thinner)
        target_node.verbose("Receive datasets under: {}".format(self.args.target_path))

        if self.args.no_send:
            self.set_title("Thinning source and target")
        else:
            self.set_title("Sending and thinning")

        # check if exists, to prevent vague errors
        target_dataset = ZfsDataset(target_node, self.args.target_path)
        if not target_dataset.exists:
            self.error("Target path '{}' does not exist. Please create this dataset first.".format(target_dataset))
            return 255

        if self.args.filter_properties:
            filter_properties = self.args.filter_properties.split(",")
        else:
            filter_properties = []

        if self.args.set_properties:
            set_properties = self.args.set_properties.split(",")
        else:
            set_properties = []

        if self.args.clear_refreservation:
            filter_properties.append("refreservation")

        if self.args.clear_mountpoint:
            set_properties.append("canmount=noauto")

        # sync datasets
        fail_count = 0
        target_datasets = []
        for source_dataset in source_datasets:

            try:
                # determine corresponding target_dataset
                target_name = self.args.target_path + "/" + source_dataset.lstrip_path(self.args.strip_path)
                target_dataset = ZfsDataset(target_node, target_name)
                target_datasets.append(target_dataset)

                # ensure parents exists
                # TODO: this isnt perfect yet, in some cases it can create parents when it shouldn't.
                if not self.args.no_send \
                        and target_dataset.parent not in target_datasets \
                        and not target_dataset.parent.exists:
                    target_dataset.parent.create_filesystem(parents=True)

                # determine common zpool features
                source_features = source_node.get_zfs_pool(source_dataset.split_path()[0]).features
                target_features = target_node.get_zfs_pool(target_dataset.split_path()[0]).features
                common_features = source_features and target_features
                # source_dataset.debug("Common features: {}".format(common_features))

                source_dataset.sync_snapshots(target_dataset, show_progress=self.args.progress,
                                              features=common_features, filter_properties=filter_properties,
                                              set_properties=set_properties,
                                              ignore_recv_exit_code=self.args.ignore_transfer_errors,
                                              holds=not self.args.no_holds, rollback=self.args.rollback,
                                              raw=self.args.raw, other_snapshots=self.args.other_snapshots,
                                              no_send=self.args.no_send,
                                              destroy_incompatible=self.args.destroy_incompatible)
            except Exception as e:
                fail_count = fail_count + 1
                source_dataset.error("FAILED: " + str(e))
                if self.args.debug:
                    raise

        # if not self.args.no_thinning:
        self.thin_missing_targets(ZfsDataset(target_node, self.args.target_path), target_datasets)

        return fail_count

    def thin_missing_targets(self, target_dataset, used_target_datasets):
        """thin/destroy target datasets that are missing on the source."""

        self.debug("Thinning obsolete datasets")

        for dataset in target_dataset.recursive_datasets:
            try:
                if dataset not in used_target_datasets:
                    dataset.debug("Missing on source, thinning")
                    dataset.thin()

                    # destroy_missing enabled?
                    if self.args.destroy_missing is not None:

                        # cant do anything without our own snapshots
                        if not dataset.our_snapshots:
                            if dataset.datasets:
                                dataset.debug("Destroy missing: ignoring")
                            else:
                                dataset.verbose(
                                    "Destroy missing: has no snapshots made by us. (please destroy manually)")
                        else:
                            # past the deadline?
                            deadline_ttl = ThinnerRule("0s" + self.args.destroy_missing).ttl
                            now = int(time.time())
                            if dataset.our_snapshots[-1].timestamp + deadline_ttl > now:
                                dataset.verbose("Destroy missing: Waiting for deadline.")
                            else:

                                dataset.debug("Destroy missing: Removing our snapshots.")

                                # remove all our snaphots, except last, to safe space in case we fail later on
                                for snapshot in dataset.our_snapshots[:-1]:
                                    snapshot.destroy(fail_exception=True)

                                # does it have other snapshots?
                                has_others = False
                                for snapshot in dataset.snapshots:
                                    if not snapshot.is_ours():
                                        has_others = True
                                        break

                                if has_others:
                                    dataset.verbose("Destroy missing: Still in use by other snapshots")
                                else:
                                    if dataset.datasets:
                                        dataset.verbose("Destroy missing: Still has children here.")
                                    else:
                                        dataset.verbose("Destroy missing.")
                                        dataset.our_snapshots[-1].destroy(fail_exception=True)
                                        dataset.destroy(fail_exception=True)

            except Exception as e:
                dataset.error("Error during destoy missing ({})".format(str(e)))

    def thin_source(self, source_datasets):

        self.set_title("Thinning source")

        for source_dataset in source_datasets:
            source_dataset.thin(skip_holds=True)

    def run(self):

        try:
            self.verbose(HEADER)

            if self.args.test:
                self.verbose("TEST MODE - SIMULATING WITHOUT MAKING ANY CHANGES")

            self.set_title("Source settings")

            description = "[Source]"
            source_thinner = Thinner(self.args.keep_source)
            source_node = ZfsNode(self.args.backup_name, self, ssh_config=self.args.ssh_config,
                                  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 "
                "'autobackup:{}=child')".format(
                    self.args.backup_name, self.args.backup_name))

            self.set_title("Selecting")
            selected_source_datasets = source_node.selected_datasets
            if not selected_source_datasets:
                self.error(
                    "No source filesystems selected, please do a 'zfs set autobackup:{0}=true' on the source datasets "
                    "you want to select.".format(
                        self.args.backup_name))
                return 255

            source_datasets = []

            # filter out already replicated stuff?
            if not self.args.ignore_replicated:
                source_datasets = selected_source_datasets
            else:
                self.set_title("Filtering already replicated filesystems")
                for selected_source_dataset in selected_source_datasets:
                    if selected_source_dataset.is_changed(self.args.min_change):
                        source_datasets.append(selected_source_dataset)
                    else:
                        selected_source_dataset.verbose("Ignoring, already replicated")

            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)

            # if target is specified, we sync the datasets, otherwise we just thin the source. (e.g. snapshot mode)
            if self.args.target_path:
                fail_count = self.sync_datasets(source_node, source_datasets)
            else:
                self.thin_source(source_datasets)
                fail_count = 0

            if not fail_count:
                if self.args.test:
                    self.set_title("All tests successfull.")
                else:
                    self.set_title("All operations completed successfully")
                    if not self.args.target_path:
                        self.verbose("(No target_path specified, only operated as snapshot tool.)")

            else:
                if fail_count != 255:
                    self.error("{} failures!".format(fail_count))

            if self.args.test:
                self.verbose("")
                self.verbose("TEST MODE - DID NOT MAKE ANY CHANGES!")

            return fail_count

        except Exception as e:
            self.error("Exception: " + str(e))
            if self.args.debug:
                raise
            return 255
        except KeyboardInterrupt:
            self.error("Aborted")
            return 255


if __name__ == "__main__":
    zfs_autobackup = ZfsAutobackup(sys.argv[1:], False)
    sys.exit(zfs_autobackup.run())