Merge pull request #221 from timvaillancourt/1.2.0-backup-retention

1.2.0: Support backup rotation (#146)

README.rst

@@ -35,6 +35,7 @@ Features
 -  `Nagios NSCA <https://sourceforge.net/p/nagios/nsca>`__ push
    notification support (*optional*)
 -  Modular backup, archiving, upload and notification components
+-  Rotation of backups by time or count
 -  Support for MongoDB Authentication and SSL database connections
 -  Multi-threaded, single executable
 -  Auto-scales to number of available CPUs by default

conf/mongodb-consistent-backup.example.conf

@@ -17,7 +17,10 @@ production:
     location: /var/lib/mongodb-consistent-backup
   #  mongodump:
   #    binary: [path]                (default: /usr/bin/mongodump)
-  #    compression: [auto|none|gzip] (default: auto = enable gzip if supported)
+  #    compression: [auto|none|gzip] (default: auto - enable gzip if supported)
+  #rotate:
+  #  max_backups: [1+]
+  #  max_days: [0.1+]
   #    threads: [1-16]               (default: auto-generated, shards/cpu)
   #replication:
   #  max_lag_secs: [1+]                        (default: 10)

mongodb_consistent_backup/Common/Config.py

@@ -73,6 +73,8 @@ def makeParser(self):
         parser.add_argument("--ssl.client_cert_file", dest="ssl.client_cert_file", help="Path to Client SSL Certificate file in PEM format (for optional client ssl auth)", default=None, type=str)
         parser.add_argument("-L", "--log-dir", dest="log_dir", help="Path to write log files to (default: disabled)", default='', type=str)
         parser.add_argument("--lock-file", dest="lock_file", help="Location of lock file (default: /tmp/mongodb-consistent-backup.lock)", default='/tmp/mongodb-consistent-backup.lock', type=str)
+        parser.add_argument("--rotate.max_backups", dest="rotate.max_backups", help="Maximum number of backups to keep in backup directory (default: unlimited)", default=0, type=int)
+        parser.add_argument("--rotate.max_days", dest="rotate.max_days", help="Maximum age in days for backups in backup directory (default: unlimited)", default=0, type=float)
         parser.add_argument("--sharding.balancer.wait_secs", dest="sharding.balancer.wait_secs", help="Maximum time to wait for balancer to stop, in seconds (default: 300)", default=300, type=int)
         parser.add_argument("--sharding.balancer.ping_secs", dest="sharding.balancer.ping_secs", help="Interval to check balancer state, in seconds (default: 3)", default=3, type=int)
         return self.makeParserLoadSubmodules(parser)

mongodb_consistent_backup/Main.py

@@ -14,8 +14,9 @@
 from Notify import Notify
 from Oplog import Tailer, Resolver
 from Replication import Replset, ReplsetSharded
+from Rotate import Rotate
 from Sharding import Sharding
-from State import StateRoot, StateBackup, StateBackupReplset, StateDoneStamp, StateOplog
+from State import StateRoot, StateBackup, StateBackupReplset, StateOplog
 from Upload import Upload
 
 
@@ -91,14 +92,13 @@ def setup_signal_handlers(self):
 
     def set_backup_dirs(self):
         self.backup_root_directory    = os.path.join(self.config.backup.location, self.config.backup.name)
-        self.backup_latest_symlink    = os.path.join(self.backup_root_directory, "latest")
-        self.backup_previous_symlink  = os.path.join(self.backup_root_directory, "previous")
         self.backup_root_subdirectory = os.path.join(self.config.backup.name, self.backup_time)
         self.backup_directory         = os.path.join(self.config.backup.location, self.backup_root_subdirectory)
 
     def setup_state(self):
-        StateRoot(self.backup_root_directory, self.config).write(True)
-        self.state = StateBackup(self.backup_directory, self.config, self.backup_time, self.uri, sys.argv)
+        self.state_root = StateRoot(self.backup_root_directory, self.config)
+        self.state      = StateBackup(self.backup_directory, self.config, self.backup_time, self.uri, sys.argv)
+        self.state_root.write(True)
         self.state.write()
 
     def setup_notifier(self):
@@ -151,21 +151,9 @@ def stop_timer(self):
         self.timer.stop(self.timer_name)
         self.state.set('timers', self.timer.dump())
 
-    def read_symlink_latest(self):
-        if os.path.islink(self.backup_latest_symlink):
-            return os.readlink(self.backup_latest_symlink)
-
-    def update_symlinks(self):
-        latest = self.read_symlink_latest()
-        if latest:
-            logging.info("Updating %s previous symlink to: %s" % (self.config.backup.name, latest))
-            if os.path.islink(self.backup_previous_symlink):
-                os.remove(self.backup_previous_symlink)
-            os.symlink(latest, self.backup_previous_symlink)
-        if os.path.islink(self.backup_latest_symlink):
-            os.remove(self.backup_latest_symlink)
-        logging.info("Updating %s latest symlink to: %s" % (self.config.backup.name, self.backup_directory))
-        return os.symlink(self.backup_directory, self.backup_latest_symlink)
+    def rotate_backups(self):
+        rotater = Rotate(self.config, self.state_root, self.state)
+        rotater.run()
 
     # TODO Rename class to be more exact as this assumes something went wrong
     # noinspection PyUnusedLocal
@@ -451,6 +439,7 @@ def run(self):
 
         # stop timer
         self.stop_timer()
+        self.state.set("completed", True)
 
         # send notifications of backup state
         try:
@@ -466,8 +455,7 @@ def run(self):
             self.notify.close()
             self.exception("Problem running Notifier! Error: %s" % e, e)
 
-        StateDoneStamp(self.backup_directory, self.config).write()
-        self.update_symlinks()
+        self.rotate_backups()
 
         self.logger.rotate()
         logging.info("Completed %s in %.2f sec" % (self.program_name, self.timer.duration(self.timer_name)))

mongodb_consistent_backup/Rotate.py

@@ -0,0 +1,97 @@
+import logging
+import os
+
+from math import ceil
+from shutil import rmtree
+from time import time
+
+from mongodb_consistent_backup.Errors import OperationError
+
+
+class Rotate(object):
+    def __init__(self, config, state_root, state_bkp):
+        self.config      = config
+        self.state_root  = state_root
+        self.state_bkp   = state_bkp
+        self.backup_name = self.config.backup.name
+        self.max_backups = self.config.rotate.max_backups
+        self.max_days    = self.config.rotate.max_days
+
+        self.previous = None
+        self.backups  = self.backups_by_unixts()
+        self.latest   = state_bkp.get()
+
+        self.base_dir         = os.path.join(self.config.backup.location, self.config.backup.name)
+        self.latest_symlink   = os.path.join(self.base_dir, "latest")
+        self.previous_symlink = os.path.join(self.base_dir, "previous")
+
+        self.max_secs = 0
+        if self.max_days > 0:
+            seconds       = float(self.max_days) * 86400.00
+            self.max_secs = int(ceil(seconds))
+
+    def backups_by_unixts(self):
+        backups = {}
+        for name in self.state_root.backups:
+            backup      = self.state_root.backups[name]
+            backup_time = backup["updated_at"]
+            backups[backup_time] = backup
+            if not self.previous or backup_time > self.previous["updated_at"]:
+                self.previous = backup
+        return backups
+
+    def remove(self, ts):
+        if ts in self.backups:
+            backup = self.backups[ts]
+            if os.path.isdir(backup["path"]):
+                logging.debug("Removing backup path: %s" % backup["path"])
+                rmtree(backup["path"])
+            else:
+                raise OperationError("Backup path %s does not exist!" % backup["path"])
+            if self.previous == backup:
+                self.previous = None
+            del self.backups[ts]
+
+    def rotate(self):
+        if self.max_days == 0 and self.max_backups == 0:
+            logging.info("Backup rotation is disabled, skipping")
+            return
+        logging.info("Rotating backups (max_backups=%i, max_days=%.2f)" % (self.max_backups, self.max_days))
+        kept_backups = 1
+        now = int(time())
+        remove_backups = {}
+        for ts in sorted(self.backups.iterkeys(), reverse=True):
+            backup = self.backups[ts]
+            name   = backup["name"].encode("ascii", "ignore")
+            if self.max_backups == 0 or kept_backups < self.max_backups:
+                if self.max_secs > 0 and (now - ts) > self.max_secs:
+                    remove_backups[name] = ts
+                    continue
+                logging.debug("Keeping previous backup %s" % name)
+                kept_backups += 1
+            else:
+                remove_backups[name] = ts
+        if len(remove_backups) > 0:
+            logging.info("Backup(s) exceeds max backup count or age, removing: %s" % sorted(remove_backups.keys()))
+            for name in remove_backups:
+                self.remove(remove_backups[name])
+
+    def symlink(self):
+        try:
+            if os.path.islink(self.latest_symlink):
+                os.remove(self.latest_symlink)
+            logging.info("Updating %s latest symlink to current backup path: %s" % (self.backup_name, self.latest["path"]))
+            os.symlink(self.latest["path"], self.latest_symlink)
+
+            if os.path.islink(self.previous_symlink):
+                os.remove(self.previous_symlink)
+            if self.previous:
+                logging.info("Updating %s previous symlink to: %s" % (self.backup_name, self.previous["path"]))
+                os.symlink(self.previous["path"], self.previous_symlink)
+        except Exception, e:
+            logging.error("Error creating backup symlinks: %s" % e)
+            raise OperationError(e)
+
+    def run(self):
+        self.rotate()
+        self.symlink()

mongodb_consistent_backup/State.py

@@ -37,10 +37,12 @@ def merge(self, new, old):
         merged.update(new)
         return merged
 
-    def load(self, load_one=False):
+    def load(self, load_one=False, filename=None):
         f = None
+        if not filename:
+            filename = self.state_file
         try:
-            f = open(self.state_file, "r")
+            f = open(filename, "r")
             data = decode_all(f.read())
             if load_one and len(data) > 0:
                 return data[0]
@@ -51,6 +53,15 @@ def load(self, load_one=False):
             if f:
                 f.close()
 
+    def get(self, key=None):
+        if key in self.state:
+            return self.state[key]
+        return self.state
+
+    def set(self, name, summary):
+        self.state[name] = summary
+        self.write(True)
+
     def write(self, do_merge=False):
         f = None
         try:
@@ -94,6 +105,7 @@ def __init__(self, base_dir, config, backup_time, seed_uri, argv=None):
         StateBase.__init__(self, base_dir, config)
         self.base_dir            = base_dir
         self.state['backup']     = True
+        self.state['completed']  = False
         self.state['name']       = backup_time
         self.state['method']     = config.backup.method
         self.state['path']       = base_dir
@@ -118,16 +130,14 @@ def __init__(self, base_dir, config, backup_time, seed_uri, argv=None):
     def init(self):
         logging.info("Initializing backup state directory: %s" % self.base_dir)
 
-    def set(self, name, summary):
-        self.state[name] = summary
-        self.write(True)
-
 
 class StateRoot(StateBase):
     def __init__(self, base_dir, config):
         StateBase.__init__(self, base_dir, config)
         self.base_dir = base_dir
         self.state['root'] = True
+        self.backups = {}
+        self.completed_backups = 0
 
         self.init()
 
@@ -136,7 +146,6 @@ def init(self):
         self.load_backups()
 
     def load_backups(self):
-        backups = []
         if os.path.isdir(self.base_dir):
             for subdir in os.listdir(self.base_dir):
                 try:
@@ -145,16 +154,10 @@ def load_backups(self):
                         continue
                     state_path = os.path.join(bkp_path, self.meta_name)
                     state_file = os.path.join(state_path, "meta.bson")
-                    done_path  = os.path.join(state_path, "done.bson")
-                    if os.path.isdir(state_path) and os.path.isfile(state_file) and os.path.isfile(done_path):
-                        backups.append(state_file)
+                    self.backups[subdir] = self.load(True, state_file)
+                    if self.backups[subdir]["completed"]:
+                        self.completed_backups += 1
                 except:
                     continue
-            logging.info("Found %i existing completed backups for set" % len(backups))
-        return backups
-
-
-class StateDoneStamp(StateBase):
-    def __init__(self, base_dir, config):
-        StateBase.__init__(self, base_dir, config, "done.bson")
-        self.state = {'done': True}
+            logging.info("Found %i existing completed backups for set" % self.completed_backups)
+        return self.backups