#!/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-beta3" 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 def error(self, 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 replaces itself with an ordinary attribute. Deleting the attribute resets the property. 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] 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.force_exists = force_exists self._cached_properties = {} 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 cache""" # TODO: nicer? self._cached_properties = {} self.force_exists = None 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: 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""" self.debug("Getting zfs properties") cmd = [ "zfs", "get", "-H", "-o", "property,value", "-p", "all", self.name ] if not self.exists: return {} 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 @CachedProperty def snapshots(self): """get all snapshots of this dataset""" self.debug("Getting snapshots") if not self.exists: return [] cmd = [ "zfs", "list", "-d", "1", "-r", "-t", "snapshot", "-H", "-o", "name", self.name ] names = self.zfs_node.run(cmd=cmd, readonly=True) return self.from_names(names) @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""" self.debug("Getting bytes written since our last snapshot") latest_snapshot = self.our_snapshots[-1] 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""" self.debug("Rolling back") for snapshot in reversed(self.snapshots): if snapshot.exists: 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, source_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) self.debug("Create thinning list") if self.our_snapshots: (source_keeps, source_obsoletes) = self.thin_list(keeps=[self.our_snapshots[-1]]) else: source_keeps = [] 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 target_snapshot.hold() if source_holds: source_snapshot.hold() if prev_source_snapshot: if source_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 snapshot = ZfsDataset(dataset.zfs_node, dataset.name + "@" + snapshot_name) pool = dataset.split_path()[0] if pool not in pools: pools[pool] = [] pools[pool].append(snapshot) # add snapshot to cache (also useful in testmode) dataset.snapshots.append(snapshot) # NOTE: this will trigger zfs list 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 lock snapshots on the source. (Useful to allow proxmox HA replication to ' 'switches nodes)') 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)") 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, source_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())