changeset
Pruning support
Wed, 31 Jan 2018 00:06:54 +0000
- author
- Tuomo Valkonen <tuomov@iki.fi>
- date
- Wed, 31 Jan 2018 00:06:54 +0000
- changeset 97
- 96d5adbe0205
- parent 96
- de8ac6c470d8
- child 98
- 9052e427ea39
Pruning support
| borgend/backup.py | file | annotate | diff | comparison | revisions | |
| borgend/config.py | file | annotate | diff | comparison | revisions | |
| borgend/dreamtime.py | file | annotate | diff | comparison | revisions | |
| borgend/instance.py | file | annotate | diff | comparison | revisions | |
| borgend/ui.py | file | annotate | diff | comparison | revisions | |
| config.example.yaml | file | annotate | diff | comparison | revisions |
--- a/borgend/backup.py Mon Jan 29 14:32:27 2018 +0000 +++ b/borgend/backup.py Wed Jan 31 00:06:54 2018 +0000 @@ -8,12 +8,13 @@ import logging import time import datetime +import re from enum import IntEnum from threading import Thread, Lock, Condition from . import config from . import repository -from . import dreamtime +from .dreamtime import MonotonicTime, DreamTime, RealTime from .instance import BorgInstance from .scheduler import TerminableThread @@ -61,25 +62,45 @@ INFO='info' LIST='list' - def __init__(self, operation, time, **kwargs): - self.operation=operation - self.time=time + def __init__(self, type, start_time, **kwargs): + self.type=type + self.start_time=start_time + self.finish_time=None self.detail=kwargs + self.errors=Errors.OK def when(self): - return self.time.realtime() + return self.start_time.realtime() + + def ok(self): + return self.errors.ok() + + def add_error(self, error): + self.errors=self.errors.combine(error) class Status(Operation): def __init__(self, backup, op=None): + op=None + errorsop=None + + if backup.current_operation: + op=backup.current_operation + elif backup.scheduled_operation: + op=backup.scheduled_operation + if backup.previous_operation: + errorsop=backup.previous_operation + if op: - super().__init__(op.operation, op.time, **op.detail) + super().__init__(op.type, op.start_time, **op.detail) + if not errorsop: + errorsop=op + self.errors=errorsop.errors else: super().__init__(None, None) self.name=backup.name self.state=backup.state - self.errors=backup.errors # # Miscellaneous helper routines @@ -139,6 +160,22 @@ return thistime, True +_prune_progress_re=re.compile(".*\(([0-9]+)/([0-9]+)\)$") +# Borg gives very little progress info in easy form, so try to extrat it +def check_prune_status(msg): + res=_prune_progress_re.match(msg) + if res: + c=res.groups() + try: + archive_no=int(c[0]) + of_total=int(c[1]) + except: + pass + else: + return archive_no, of_total + return None, None + + # # The Backup class # @@ -173,6 +210,10 @@ 'Backup interval', loc, config.defaults['backup_interval']) + self.prune_interval=config.check_nonneg_int(cfg, 'prune_interval', + 'Prune interval', loc, + config.defaults['prune_interval']) + self.retry_interval=config.check_nonneg_int(cfg, 'retry_interval', 'Retry interval', loc, config.defaults['retry_interval']) @@ -183,9 +224,9 @@ default="dreamtime") if scheduling=="dreamtime": - self.timeclass=dreamtime.DreamTime + self.timeclass=DreamTime elif scheduling=="realtime": - self.timeclass=dreamtime.MonotonicTime + self.timeclass=MonotonicTime elif scheduling=="manual": self.backup_interval=0 else: @@ -206,14 +247,12 @@ self.thread_log=None self.thread_res=None + self.previous_operation=None self.current_operation=None self.scheduled_operation=None - self.last_operation_finished=None - self.last_create_when=None - self.last_create_finished=None + self.previous_operation_of_type={} self.state=State.INACTIVE - self.errors=Errors.OK - self.timeclass=dreamtime.DreamTime + self.timeclass=DreamTime self.__decode_config(cfg) @@ -241,7 +280,7 @@ self.logger.debug('Log listener thread waiting for entries') success=True for msg in iter(self.borg_instance.read_log, None): - self.logger.info(str(msg)) + self.logger.debug(str(msg)) t=msg['type'] errormsg=None @@ -293,13 +332,20 @@ msg['msgid']=='LockErrorT')): # in docs errors=Errors.BUSY with self._cond: - self.errors=self.errors.combine(errors) + self.current_operation.add_error(errors) + status, callback=self.__status_unlocked() + elif lvl==logging.INFO and self.current_operation.type==Operation.PRUNE: + # Borg gives very little progress info in easy form, so try to extrat it + archive_number, of_total=check_prune_status(msg['message']) + if archive_number!=None and of_total!=None: + self.current_operation.detail['progress_current_secondary']=archive_number + self.current_operation.detail['progress_total_secondary']=of_total status, callback=self.__status_unlocked() elif t=='question_prompt' or t=='question_prompt_retry': self.logger.error('Did not expect to receive question prompt from borg') with self._cond: - self.errors=self.errors.combine(Errors.ERRORS) + self.current_operation.add_error(Errors.ERRORS) # TODO: terminate org? Send 'NO' reply? elif (t=='question_invalid_answer' or t=='question_accepted_default' @@ -328,14 +374,17 @@ if res is None: with self._cond: - if self.errors.ok(): + # Prune gives absolutely no result, so don't complain + if (self.current_operation.ok() and + self.current_operation.type!=Operation.PRUNE): self.logger.error('No result from borg despite no error in log') - self.errors=Errors.ERRORS - elif self.current_operation.operation==Operation.CREATE: - with self._cond: - self.last_create_finished=time.monotonic() - self.last_create_when=self.current_operation.time.monotonic() - elif self.current_operation.operation==Operation.LIST: + self.current_operation.add_error(Errors.ERRORS) + elif self.current_operation.type==Operation.LIST: + self.__handle_list_result(res) + + # All other results are discarded + + def __handle_list_result(self, res): ok=True latest=None if 'archives' in res and isinstance(res['archives'], list): @@ -354,27 +403,24 @@ if not ok: with self._cond: - self.errors=self.errors.combine(Errors.ERRORS) + self.current_operation.add_error(Errors.ERRORS) if latest: self.logger.info('borg info: Previous backup was on %s' % latest.isoformat()) - realtime=time.mktime(latest.timetuple()) - monotonic=realtime+time.monotonic()-time.time() + when=MonotonicTime.from_realtime(time.mktime(latest.timetuple())) + op=Operation(Operation.CREATE, when, reason='listed') + op.finish_time=when with self._cond: - self.last_create_finished=monotonic - self.last_create_when=monotonic + self.previous_operation_of_type[Operation.CREATE]=op else: self.logger.info('borg info: Could not discover a previous backup') - with self._cond: - self.last_create_finished=-1 - self.last_create_when=None def __do_launch(self, op, archive_or_repository, common_params, op_params, paths=[]): self.logger.debug('Creating BorgInstance') - inst=BorgInstance(op.operation, archive_or_repository, + inst=BorgInstance(op.type, archive_or_repository, common_params, op_params, paths) self.logger.debug('Launching BorgInstance via repository') @@ -396,10 +442,9 @@ self.current_operation=op # Update scheduled time to real starting time to schedule # next run relative to this - self.current_operation.time=dreamtime.MonotonicTime.now() + self.current_operation.start_time=MonotonicTime.now() self.state=State.ACTIVE # Reset error status when starting a new operation - self.errors=Errors.OK self.__update_status() t_log.start() @@ -407,28 +452,28 @@ def __launch(self, op): - self.logger.debug("Launching '%s'" % str(op.operation)) + self.logger.debug("Launching '%s'" % str(op.type)) params=(config.borg_parameters +self.repository.borg_parameters +self.borg_parameters) - if op.operation==Operation.CREATE: + if op.type==Operation.CREATE: archive="%s::%s%s" % (self.repository.location, self.archive_prefix, self.archive_template) self.__do_launch(op, archive, params.common, params.create, self.paths) - elif op.operation==Operation.PRUNE: + elif op.type==Operation.PRUNE: self.__do_launch(op, self.repository.location, params.common, - [{'prefix': self.archive_prefix}] + params.create) + [{'prefix': self.archive_prefix}] + params.prune) - elif op.operation==Operation.LIST: + elif op.type==Operation.LIST: self.__do_launch(op, self.repository.location, params.common, [{'prefix': self.archive_prefix}]) else: - raise NotImplementedError("Invalid operation '%s'" % str(op.operation)) + raise NotImplementedError("Invalid operation '%s'" % str(op.type)) # This must be called with self._cond held. def __launch_and_wait(self): @@ -443,6 +488,8 @@ self.__wait_finish() def __wait_finish(self): + current=self.current_operation + # Wait for main logger thread to terminate, or for us to be terminated while not self.terminate and self.thread_res.is_alive(): self._cond.release() @@ -465,14 +512,16 @@ if not self.borg_instance.wait(): self.logger.error('Borg subprocess did not terminate') - self.errors=self.errors.combine(Errors.ERRORS) + curent.add_error(Errors.ERRORS) + + current.finish_time=MonotonicTime.now() - now=time.monotonic() - self.last_operation_finished=now + self.previous_operation_of_type[current.type]=current + self.previous_operation=current + self.current_operation=None self.thread_res=None self.thread_log=None self.borg_instance=None - self.current_operation=None self.state=State.INACTIVE self.__update_status() @@ -525,17 +574,17 @@ if not self.scheduled_operation: op=self.__next_operation_unlocked() if op: - self.logger.info("Scheduling '%s' (detail: %s) in %d seconds [%s]" % - (str(op.operation), op.detail or 'none', - op.time.seconds_to(), - op.time.__class__.__name__)) + self.logger.info("Scheduling '%s' (detail: %s) on %s [%s]" % + (str(op.type), op.detail or 'none', + op.start_time.isoformat(), + op.start_time.__class__.__name__)) self.scheduled_operation=op self.state=State.SCHEDULED self.__update_status() # Wait under scheduled wait - self.scheduler.wait_until(op.time, self._cond, self.backup_name) + self.scheduler.wait_until(op.start_time, self._cond, self.backup_name) else: # Nothing scheduled - just wait self.logger.info("Waiting for manual scheduling") @@ -564,52 +613,93 @@ self.__update_status() def __next_operation_unlocked(self): - if self.backup_interval==0: - # Manual backup + listop=self.__next_operation_list() + if listop: + return listop + + create=self.__next_operation_type(Operation.CREATE, + self.backup_interval, + important=True, + initial_reason='initial'); + + prune=self.__next_operation_type(Operation.PRUNE, + self.prune_interval, + important=False, + initial_reason=None); + + if not prune: + return create + elif not create: + return prune + elif create.start_time < prune.start_time: + return create + else: + return prune + + def __next_operation_list(self): + reason='initial' + # Unless manual backup has been chosen (backup_interval<=0), perform + # repository listing if no previous create operation known, or if we + # just pruned the repository + if self.backup_interval<=0: return None - elif self.last_create_finished==None: - # Don't know when last create finished: try to get it from - # archive list. - if not self.errors.ok() and self.retry_interval==0: + elif (self.previous_operation and + self.previous_operation.type==Operation.PRUNE): + tm=MonotonicTime.now() + reason='post-prune' + elif Operation.LIST in self.previous_operation_of_type: + prev=self.previous_operation_of_type[Operation.LIST] + if prev.ok(): + return None + if self.retry_interval==0: # Do not retry in case of errors if retry interval is zero return None - elif self.last_operation_finished: - # Attempt ater retry interval if some operation has been run - # already - tm=dreamtime.MonotonicTime(self.last_operation_finished+self.retry_interval) - else: - # Nothing has been attempted: run immediately - tm=dreamtime.MonotonicTime.now() - return Operation(Operation.LIST, tm, reason='initial') - elif self.errors.ok() and self.last_create_finished<0: - # No backup exists; perform initial backup - tm=dreamtime.MonotonicTime.now() - return Operation(Operation.CREATE, tm, reason='initial') - elif not self.errors.ok(): - # Retry create in case of errors unless retry has been disabled - if self.retry_interval==0: + # Attempt after retry interval + tm=MonotonicTime.after_other(prev.finish_time, self.retry_interval) + else: + # Nothing has been attempted: run immediately + tm=MonotonicTime.now() + return Operation(Operation.LIST, tm, reason=reason) + + def __next_operation_type(self, optype, standard_interval, + important=False, + initial_reason=None): + if optype not in self.previous_operation_of_type: + # No previous operation exists; perform immediately + # if important, otherwise after standard interval. + # Do not perform if manual operation selected by + # setting standard_interval<=0 + if standard_interval<=0: return None else: - if self.last_create_finished<0: - basetime=time.monotonic() + if important: + tm=MonotonicTime.now() + else: + tm=self.timeclass.after(standard_interval) + if initial_reason: + return Operation(optype, tm, reason=initial_reason) else: - basetime=self.last_create_finished - tm=dreamtime.MonotonicTime(basetime+self.retry_interval) - return Operation(Operation.CREATE, tm, reason='retry') + return Operation(optype, tm) else: - # All ok - run create at standard backup interval - tm=self.timeclass.from_monotonic(self.last_create_when+self.backup_interval) - return Operation(Operation.CREATE, tm) + # Previous operation has been performed; perform after + # retry interval if there were errors, otherwise after + # standard interval. + prev=self.previous_operation_of_type[optype] + if not prev.ok(): + tm=MonotonicTime.after_other(prev.start_time, + self.retry_interval) + return Operation(optype, tm, reason='retry') + elif standard_interval>0: + tm=self.timeclass.after_other(prev.start_time, + standard_interval) + return Operation(optype, tm) + else: + # Manual operation is standard_interval is zero. + return None def __status_unlocked(self): callback=self.__status_update_callback - - if self.current_operation: - status=Status(self, self.current_operation) - elif self.scheduled_operation: - status=Status(self, self.scheduled_operation) - else: - status=Status(self) + status=Status(self) return status, callback @@ -638,13 +728,13 @@ return res[0] def create(self): - op=Operation(Operation.CREATE, dreamtime.MonotonicTime.now(), reason='manual') + op=Operation(Operation.CREATE, MonotonicTime.now(), reason='manual') with self._cond: self.scheduled_operation=op self._cond.notify() def prune(self): - op=Operation(Operation.PRUNE, dreamtime.MonotonicTime.now(), reason='manual') + op=Operation(Operation.PRUNE, MonotonicTime.now(), reason='manual') with self._cond: self.scheduled_operation=op self._cond.notify()
--- a/borgend/config.py Mon Jan 29 14:32:27 2018 +0000 +++ b/borgend/config.py Wed Jan 31 00:06:54 2018 +0000 @@ -29,6 +29,8 @@ 'backup_interval': 21600, # Default: retry every 15 minutes if unable to connect / unfinished backup 'retry_interval': 900, + # Default: do not prune (0) + 'prune_interval': 0, # Extract passphrases at startup or on demand? 'extract_passphrases_at_startup': True, # Do not insert a quit menu entry (useful for installing on computers of
--- a/borgend/dreamtime.py Mon Jan 29 14:32:27 2018 +0000 +++ b/borgend/dreamtime.py Wed Jan 31 00:06:54 2018 +0000 @@ -31,6 +31,9 @@ return max(0, time.monotonic()-dreamtime_difference()) class Time: + def __init__(self, when): + self._value=when + def realtime(self): raise NotImplementedError @@ -57,12 +60,22 @@ def after(cls, seconds): return cls(cls._now()+seconds) + @classmethod + def after_other(cls, other, seconds): + if isinstance(other, cls): + return cls(other._value+seconds) + else: + return cls.from_monotonic(other.monotonic()+seconds) + def datetime(self): return datetime.datetime.fromtimestamp(self.realtime()) def seconds_to(self): return self._value-self._now() + def isoformat(self): + return self.datetime().isoformat() + def __lt__(self, other): return self.monotonic() < other.monotonic() @@ -79,9 +92,6 @@ return self.monotonic() == other.realtime() class RealTime(Time): - def __init__(self, when): - self._value=when - def realtime(self): return self._value @@ -93,9 +103,6 @@ return time.time() class MonotonicTime(Time): - def __init__(self, when): - self._value=when - def realtime(self): return self._value+(time.time()-time.monotonic()) @@ -107,9 +114,6 @@ return time.monotonic() class DreamTime(Time): - def __init__(self, when): - self._value=when - def realtime(self): return self._value+(time.time()-dreamtime()) @@ -123,6 +127,7 @@ def _now(): return dreamtime() + if platform.system()=='Darwin': import Foundation
--- a/borgend/instance.py Mon Jan 29 14:32:27 2018 +0000 +++ b/borgend/instance.py Wed Jan 31 00:06:54 2018 +0000 @@ -19,6 +19,7 @@ 'create': ['--json'], 'info': ['--json'], 'list': ['--json'], + 'prune': ['--list'], } # Conversion of config into command line @@ -27,7 +28,19 @@ if args is None: return [] else: - return flatten([['--' + key, str(d[key])] for d in args for key in d]) + # Insert --key=str(value) for 'key: value' in the config. + # Boolean values are handled different, since borg does not take + # --key=true type of arguments. If the value is true --key is inserted, + # otherwise not. + def mkarg(key, value): + if isinstance(value, bool): + if value: + return ['--' + key] + else: + return [] + else: + return ['--' + key, str(value)] + return flatten([mkarg(key, d[key]) for d in args for key in d]) class BorgInstance: def __init__(self, operation, archive_or_repository, @@ -90,7 +103,7 @@ try: return json.loads(line.decode()) except Exception: - logger.exception('JSON parse failed on: %s' % str(line)) + logger.exception('JSON parse failed on stdout: %s' % str(line)) return None def read_log(self): @@ -117,7 +130,7 @@ res['type']='UNKNOWN' return res except: - logger.exception('JSON parse failed on: %s' % str(line)) + logger.exception('JSON parse failed on stderr: %s' % str(line)) errmsg=line for line in iter(stream.readline, b''):
--- a/borgend/ui.py Mon Jan 29 14:32:27 2018 +0000 +++ b/borgend/ui.py Wed Jan 31 00:06:54 2018 +0000 @@ -79,6 +79,29 @@ elif TB <= B: return '{0:.2f}TB'.format(B/TB) + +def progress_percentage(done, total, d): + progress='' + try: + percentage = 100*float(done)/float(total) + progress=': %d%%' % int(round(percentage)) + if 'operation_no' in d: + progress=':#%d%s' % (d['operation_no'], progress) + except: + pass + return progress + +def progress_parts(done, total, d): + progress='' + try: + progress=': %d/%d' % (int(done), int(total)) + if 'operation_no' in d: + progress=':#%d%s' % (d['operation_no'], progress) + except: + pass + return progress + + def make_title(status): def add_info(info, new): if info: @@ -125,26 +148,28 @@ if 'reason' in status.detail: this_info=status.detail['reason'] + ' ' - when_how_sched= "%s%s %s" % (this_info, status.operation, whenstr) + when_how_sched= "%s%s %s" % (this_info, status.type, whenstr) info=add_info(info, when_how_sched) elif status.state==backup.State.QUEUED: - info=add_info(info, "queued") + info=add_info(info, "queued %s" % status.type) elif status.state==backup.State.ACTIVE: # Operation running progress='' d=status.detail if 'progress_current' in d and 'progress_total' in d: - percentage = 100*float(d['progress_current'])/float(d['progress_total']) - progress=': %d%%' % int(round(percentage)) - if 'operation_no' in d: - progress='/#%d%s' % (d['operation_no'], progress) + progress=progress_percentage(d['progress_current'], + d['progress_total'], d) + elif ('progress_current_secondary' in d and + 'progress_total_secondary' in d): + progress=progress_parts(d['progress_current_secondary'], + d['progress_total_secondary'], d) elif 'original_size' in d and 'deduplicated_size' in d: progress=' %s→%s' % (humanbytes(d['original_size']), humanbytes(d['deduplicated_size'])) - howrunning = "running %s%s" % (status.operation, progress) + howrunning = "running %s%s" % (status.type, progress) info=add_info(info, howrunning) else:
--- a/config.example.yaml Mon Jan 29 14:32:27 2018 +0000 +++ b/config.example.yaml Wed Jan 31 00:06:54 2018 +0000 @@ -9,8 +9,18 @@ create_parameters: - exclude-from: $HOME/lib/borg-exclude-patterns.txt prune_parameters: - - daily: 7 - - weekly: 50 + # Set borg pruning parameters. The pruning interval is configured + # in each separate backup. NOTE: Pruning parameters can also be + # configured separately for each backup. + # + # Keep a daily archive for the last week + - keep-daily: 7 + # Keep a weely archive for the last year + - keep-weekly: 52 + # Keep a montly archive forever + - keep-monthly: -1 + # Set to true to enable a dry-run + - dry-run: False # Repositories: configure here locations, passphrases, # and general access parameters @@ -35,16 +45,20 @@ backups: # 1. Most files in $HOME to ssh://myserver.invalid - name: Home to 'myserver' + # Scheduling mode: dreamtime/realtime/manual + # Dreamtime scheduling discounts system sleep periods. + scheduling: dreamtime # Backup every 24 hours backup_interval: 86400 # Retry every 15 minutes if unable to connect / unfinished backup retry_interval: 900 - # Scheduling mode: dreamtime/realtime/manual - # Dreamtime scheduling discounts system sleep periods. - scheduling: dreamtime + # Prune interval: once every two weeks + prune_interval: 1209600 + # Repository and archive configuration repository: myserver archive_prefix: 'all@mylaptop-' archive_template: '{now:%Y-%m-%d_%H:%M:%S}' + # Files to include paths: - $HOME create_parameters: @@ -57,13 +71,15 @@ # 2. A subset of files $HOME more frequently to ssh://myserver.invalid - name: Work to 'myserver' + # Scheduling mode: dreamtime/realtime/manual + scheduling: dreamtime # Backup every 3 hours backup_interval: 10800 # Retry every 15 minutes if unable to connect / unfinished backup retry_interval: 900 - # Scheduling mode: dreamtime/realtime/manual - # Dreamtime scheduling discounts system sleep periods. - scheduling: dreamtime + # Prune interval: once every two weeks + prune_interval: 1209600 + # Repository and archive configuration repository: myserver archive_prefix: 'work@mylaptop-' archive_template: '{now:%Y-%m-%d_%H:%M:%S}' @@ -76,6 +92,7 @@ scheduling: manual backup_interval: 0 retry_interval: 0 + prune_interval: 0 repository: backup1 archive_prefix: 'mylaptop-' archive_template: '{now:%Y-%m-%d_%H:%M:%S}'
