Merge pull request #107 from timvaillancourt/MCB_1.0-zbackup

timvaillancourt · web-flow · commit 0805ff15ed3d · 2017-04-24T15:33:13.000+02:00
MCB_1.0: ZBackup support
diff --git a/README.rst b/README.rst
@@ -15,13 +15,15 @@ Features
 -  Auto-discovers healthy members for backup by considering replication
    lag, replication 'priority' and by preferring 'hidden' members.
 -  Creates cluster-consistent backups across many separate shards
--  Archives and compresses backups (*inline compression with mongodump
-   3.2+*)
 -  Transparent restore process (*just add --oplogReplay flag to your
    mongorestore command*)
+-  Archiving and compression of backups
+-  Block de-duplication and optional AES encryption at rest via `ZBackup <http://zbackup.org/>`__
+   archiving method
 -  AWS S3 Secure/HTTPS Multipart backup uploads (*optional*)
 -  `Nagios NSCA <https://sourceforge.net/p/nagios/nsca>`__ push
    notification support (*optional*)
+-  Modular backup, archiving, upload and notification components
 -  Multi-threaded, single executable
 
 Current Limitations
@@ -137,6 +139,36 @@ Run as Docker Container (Experimental)
     $ make docker
     $ docker run -t mongodb_consistent_backup <mongodb_consistent_backup-flags>
 
+
+ZBackup Archiving (Optional)
+~~~~~~~
+
+`ZBackup <http://zbackup.org/>`__ *(with LZMA compression)* is an optional archive method for mongodb_consistent_backup. This archive method significantly reduces disk usage for backups via deduplication and compression. 
+
+ZBackup offers block de-duplication and compression of backups and optionally supports AES-128 encryption at rest. The ZBackup archive method causes backups to be stored via ZBackup at archive time.
+
+To enable, ZBackup must be installed on your system and the 'archive.method' config file variable *(or --archive.method flag=)* must be set to 'zbackup'. Backups get stored in a repository directory named *mongodb_consistent_backup-zbackup* and must be restored using a 'zbackup restore' command.
+
+**Install on CentOS/RHEL**
+
+::
+
+    $ yum install zbackup
+
+**Install on Debian/Ubuntu**
+
+::
+
+    $ apt-get install zbackup
+
+ZBackup data is stored in a repository directory named *mongodb_consistent_backup-zbackup* and must be restored using a 'zbackup restore ...' command.
+
+**Get Backup from ZBackup Repo**
+
+::
+
+    $ zbackup restore --password-file /etc/zbackup.passwd /mnt/backup/default/mongodb_consistent_backup-zbackup/backups/20170424_0000.tar
+    
 Roadmap
 ~~~~~~~
 
diff --git a/conf/mongodb-consistent-backup.example.conf b/conf/mongodb-consistent-backup.example.conf
@@ -32,6 +32,12 @@ production:
   #  tar:
   #    compression: [none|gzip] (default: gzip, none if backup is compressed)
   #    threads: [1+]            (default: 1 per CPU)
+  #  zbackup:
+  #    binary: [path]        (default: /usr/bin/zbackup)
+  #    cache_mb: [mb]        (default: 128)
+  #    compression: [lzma]   (only lzma is supported)
+  #    password_file: [path] (default: none)
+  #    threads: [1+]         (default: 1 per CPU)
   notify:
     method: none
   #  nsca:
diff --git a/mongodb_consistent_backup/Archive/Archive.py b/mongodb_consistent_backup/Archive/Archive.py
@@ -1,4 +1,5 @@
 from mongodb_consistent_backup.Archive.Tar import Tar
+from mongodb_consistent_backup.Archive.Zbackup import Zbackup
 from mongodb_consistent_backup.Pipeline import Stage
 
 
diff --git a/mongodb_consistent_backup/Archive/Zbackup/Zbackup.py b/mongodb_consistent_backup/Archive/Zbackup/Zbackup.py
@@ -0,0 +1,175 @@
+import os
+import logging
+
+from select import select
+from subprocess import Popen, PIPE, call
+
+from mongodb_consistent_backup.Errors import OperationError
+from mongodb_consistent_backup.Pipeline import Task
+
+
+class Zbackup(Task):
+    def __init__(self, manager, config, timer, base_dir, backup_dir, **kwargs):
+        super(Zbackup, self).__init__(self.__class__.__name__, manager, config, timer, base_dir, backup_dir, **kwargs)
+        self.backup_name         = self.config.backup.name
+        self.backup_time         = os.path.basename(self.backup_dir)
+        self.zbackup_binary      = self.config.archive.zbackup.binary
+        self.zbackup_cache_mb    = self.config.archive.zbackup.cache_mb
+        self.zbackup_passwd_file = self.config.archive.zbackup.password_file
+
+        if self.config.archive.zbackup.threads and self.config.archive.zbackup.threads > 0:
+            self.threads(self.config.archive.zbackup.threads)
+
+        # only lzma compression supported (for now)
+        self.compression_method    = 'lzma'
+        self.compression_supported = ['lzma']
+
+        self.zbackup_dir         = os.path.join(self.config.backup.location, self.backup_name, "mongodb-consistent-backup_zbackup")
+        self.zbackup_backups     = os.path.join(self.zbackup_dir, "backups")
+        self.zbackup_backup_path = os.path.join(self.zbackup_backups, "%s.tar" % self.backup_time)
+        self.zbackup_bundles     = os.path.join(self.zbackup_dir, "bundles")
+        self.zbackup_info        = os.path.join(self.zbackup_dir, "info")
+        self.backup_meta_dir     = "mongodb-consistent-backup_META"
+
+        self.encrypted = False
+        self._zbackup  = None
+        self._tar      = None
+        self._version  = None
+
+        self.init()
+
+    def is_zbackup_init(self):
+        if os.path.isfile(self.zbackup_info) and os.path.isdir(self.zbackup_backups) and os.path.isdir(self.zbackup_bundles):
+            return True
+        return False
+
+    def init(self):
+        if os.path.isdir(self.zbackup_dir):
+            if self.is_zbackup_init():
+                logging.info("Found existing ZBackup storage dir at: %s (encrypted: %s)" % (self.zbackup_dir, self.encrypted))
+            else:
+                raise OperationError("ZBackup dir: %s is not a zbackup storage directory!" % self.zbackup_dir)
+        else:
+            try:
+                cmd_line = [self.zbackup_binary]
+                if self.zbackup_passwd_file:
+                    cmd_line.extend(["--password-file", self.zbackup_passwd_file, "init", self.zbackup_dir])
+                    logging.info("Using ZBackup AES encryption with password file: %s" % self.zbackup_passwd_file)
+                    self.encrypted = True
+                else:
+                    cmd_line.extend(["--non-encrypted", "init", self.zbackup_dir])
+                logging.warning("Initializing new ZBackup storage directory at: %s (encrypted: %s)" % (self.zbackup_dir, self.encrypted))
+                logging.debug("Using ZBackup command: '%s'" % cmd_line)
+                exit_code = call(cmd_line)
+                if exit_code != 0:
+                    raise OperationError("ZBackup initialization failed! Exit code: %i" % exit_code)
+            except Exception, e:
+                raise OperationError("Error creating ZBackup storage directory! Error: %s" % e)
+
+    def version(self):
+        if self._version:
+            return self._version
+        else:
+            try:
+                cmd = Popen([self.zbackup_binary, "--help"], stderr=PIPE)
+                stdout, stderr = cmd.communicate()
+                if stderr:
+                    line = stderr.split("\n")[0]
+                    if line.startswith("ZBackup") and "version " in line:
+                        fields  = line.split(" ")
+                        version = fields[len(fields) - 1]
+                        if len(version.split(".")) == 3:
+                            self._version = version
+                            return self._version
+                return None
+            except OSError, e:
+                return None
+            except Exception, e:
+                raise OperationError("Could not gather ZBackup version: %s" % e)
+    
+    def has_zbackup(self):
+        if self.version():
+           return True
+        return False
+
+    def close(self, exit_code=None, frame=None):
+        del exit_code
+        del frame
+        if not self.stopped:
+            if self._zbackup and self._zbackup.poll() == None:
+                logging.debug("Stopping running ZBackup command")
+                self._zbackup.terminate()
+            if self._tar and self._tar.poll() == None:
+                logging.debug("Stopping running ZBackup tar command")
+                self._tar.terminate()
+            self.stopped = True
+
+    def poll(self, timeout=1):
+        try:
+            poll = select([self._zbackup.stderr.fileno()], [], [], timeout)
+        except ValueError:
+            return
+        if len(poll) >= 1:
+            for fd in poll[0]:
+                line = self._zbackup.stderr.readline()
+                if line:
+                    logging.info(line.rstrip())
+
+    def wait(self):
+        try:
+            tar_done = False
+            while self._zbackup.stderr and self._tar.stderr:
+                self.poll()
+                if tar_done:
+                    self._zbackup.communicate()
+                    if self._zbackup.poll() != None:
+                        logging.info("ZBackup completed successfully with exit code: %i" % self._zbackup.returncode)
+                        if self._zbackup.returncode != 0:
+                            raise OperationError("ZBackup exited with code: %i!" % self._zbackup.returncode)
+                        break
+                elif self._tar.poll() != None:
+                    if self._tar.returncode == 0:
+                        logging.debug("ZBackup tar command completed successfully with exit code: %i" % self._tar.returncode)
+                        tar_done = True
+                    else:
+                        raise OperationError("ZBackup archiving failed on tar command with exit code: %i" % self._tar.returncode)
+        except Exception, e:
+            raise OperationError("Error reading ZBackup output: %s" % e)
+
+    def get_commands(self, base_dir, sub_dir):
+        tar          = ["tar", "--remove-files", "-C", base_dir, "-c", sub_dir]
+        zbackup      = [self.zbackup_binary, "--cache-size", "%imb" % self.zbackup_cache_mb, "--compression", self.compression()]
+        zbackup_path = os.path.join(self.zbackup_backups, "%s.%s.tar" % (self.backup_time, sub_dir))
+        if self.encrypted:
+            zbackup.extend(["--password-file", self.zbackup_passwd_file, "backup", zbackup_path])
+        else:
+            zbackup.extend(["--non-encrypted", "backup", zbackup_path])
+        return tar, zbackup
+
+    def run(self):
+        if self.has_zbackup():
+            try:
+                logging.info("Starting ZBackup version: %s (options: compression=%s, encryption=%s, threads=%i, cache_mb=%i)" %
+                    (self.version(), self.compression(), self.encrypted, self.threads(), self.zbackup_cache_mb)
+                )
+                self.running = True
+                try:
+                    for sub_dir in os.listdir(self.backup_dir):
+                        if sub_dir == self.backup_meta_dir:
+                            continue
+                        logging.info("Running ZBackup for path: %s" % os.path.join(self.backup_dir, sub_dir))
+                        tar_cmd, zbkp_cmd = self.get_commands(self.backup_dir, sub_dir)
+                        logging.debug("Running ZBackup tar command: %s" % tar_cmd)
+                        logging.debug("Running ZBackup command: %s" % zbkp_cmd)
+                        self._zbackup = Popen(zbkp_cmd, stdin=PIPE, stderr=PIPE)
+                        self._tar     = Popen(tar_cmd, stdout=self._zbackup.stdin, stderr=PIPE)
+                        self.wait()
+                except Exception, e:
+                    raise OperationError("Could not execute ZBackup: %s" % e)
+                logging.info("Completed running all ZBackups")
+                self.completed = True
+            finally:
+                self.running = False
+                self.stopped = True
+        else:
+            raise OperationError("Cannot find ZBackup at %s!" % self.zbackup_binary)
diff --git a/mongodb_consistent_backup/Archive/Zbackup/__init__.py b/mongodb_consistent_backup/Archive/Zbackup/__init__.py
@@ -0,0 +1,10 @@
+from Zbackup import Zbackup
+
+
+def config(parser):
+    parser.add_argument("--archive.zbackup.binary", dest="archive.zbackup.binary", help="Path to ZBackup binary (default: /usr/bin/zbackup)", default='/usr/bin/zbackup', type=str)
+    parser.add_argument("--archive.zbackup.cache_mb", dest="archive.zbackup.cache_mb", help="Megabytes of RAM to use as a cache for ZBackup (default: 128)", default=128, type=int)
+    parser.add_argument("--archive.zbackup.compression", dest="archive.zbackup.compression", help="Type of compression to use with ZBackup (default: lzma)", default='lzma', choices=['lzma'], type=str)
+    parser.add_argument("--archive.zbackup.password_file", dest="archive.zbackup.password_file", help="Path to ZBackup backup password file, enables AES encryption (default: none)", default=None, type=str)
+    parser.add_argument("--archive.zbackup.threads", dest="archive.zbackup.threads", help="Number of threads to use for ZBackup (default: 1-per-CPU)", default=0, type=int)
+    return parser
diff --git a/mongodb_consistent_backup/Archive/__init__.py b/mongodb_consistent_backup/Archive/__init__.py
@@ -2,5 +2,5 @@
 
 
 def config(parser):
-    parser.add_argument("--archive.method", dest="archive.method", help="Archiver method (default: tar)", default='tar', choices=['tar','none'])
+    parser.add_argument("--archive.method", dest="archive.method", help="Archiver method (default: tar)", default='tar', choices=['tar','zbackup','none'])
     return parser
diff --git a/mongodb_consistent_backup/Backup/Mongodump/Mongodump.py b/mongodb_consistent_backup/Backup/Mongodump/Mongodump.py
@@ -26,12 +26,13 @@ def __init__(self, manager, config, timer, base_dir, backup_dir, replsets, shard
         self.replsets           = replsets
         self.sharding           = sharding
 
-        self.version        = 'unknown'
-        self.threads_max    = 16
-        self.config_replset = False
-        self.dump_threads   = []
-        self.states         = {}
-        self._summary       = {}
+        self.compression_supported = ['none', 'gzip']
+        self.version               = 'unknown'
+        self.threads_max           = 16
+        self.config_replset        = False
+        self.dump_threads          = []
+        self.states                = {}
+        self._summary              = {}
 
         if self.config.backup.mongodump.threads and self.config.backup.mongodump.threads > 0:
             self.threads(self.config.backup.mongodump.threads)
@@ -162,12 +163,14 @@ def run(self):
         return self._summary
 
     def close(self):
-        logging.info("Stopping all mongodump threads")
-        if len(self.dump_threads) > 0:
-            for thread in self.dump_threads:
-                thread.terminate()
-        try:
-            self.timer.stop(self.timer_name)
-        except:
-            pass
-        logging.info("Stopped all mongodump threads")
+        if not self.stopped:
+            logging.info("Stopping all mongodump threads")
+            if len(self.dump_threads) > 0:
+                for thread in self.dump_threads:
+                    thread.terminate()
+            try:
+                self.timer.stop(self.timer_name)
+            except:
+                pass
+            logging.info("Stopped all mongodump threads")
+            self.stopped = True
diff --git a/mongodb_consistent_backup/Oplog/Resolver/Resolver.py b/mongodb_consistent_backup/Oplog/Resolver/Resolver.py
@@ -30,8 +30,9 @@ def __init__(self, manager, config, timer, base_dir, backup_dir, tailed_oplogs,
         self.tailed_oplogs = tailed_oplogs
         self.backup_oplogs = backup_oplogs
 
-        self.resolver_summary = {}
-        self.resolver_state   = {}
+        self.compression_supported = ['none', 'gzip']
+        self.resolver_summary      = {}
+        self.resolver_state        = {}
 
         try:
             self._pool = Pool(processes=self.threads(None, 2))
diff --git a/mongodb_consistent_backup/Oplog/Tailer/Tailer.py b/mongodb_consistent_backup/Oplog/Tailer/Tailer.py
@@ -23,8 +23,9 @@ def __init__(self, manager, config, timer, base_dir, backup_dir, replsets):
         self.status_secs = self.config.oplog.tailer.status_interval
         self.replsets    = replsets
 
-        self.shards   = {}
-        self._summary = {}
+        self.compression_supported = ['none', 'gzip']
+        self.shards                = {}
+        self._summary              = {}
 
     def summary(self):
         return self._summary
diff --git a/mongodb_consistent_backup/Pipeline/Task.py b/mongodb_consistent_backup/Pipeline/Task.py
@@ -23,16 +23,17 @@ def __init__(self, task_name, manager, config, timer, base_dir, backup_dir, **kw
         self.completed = False
         self.exit_code = 255
 
-        self.thread_count       = None
-        self.cpu_count          = cpu_count()
-        self.compression_method = 'none'
-        self.timer_name         = self.__class__.__name__
+        self.thread_count          = None
+        self.cpu_count             = cpu_count()
+        self.compression_method    = 'none'
+        self.compression_supported = ['none']
+        self.timer_name            = self.__class__.__name__
 
         signal(SIGINT, SIG_IGN)
         signal(SIGTERM, self.close)
 
     def compression(self, method=None):
-        if method:
+        if method and method in self.compression_supported:
             self.compression_method = parse_method(method)
             logging.info("Setting %s compression method: %s" % (self.task_name, self.compression_method))
         return parse_method(self.compression_method)
diff --git a/mongodb_consistent_backup/State.py b/mongodb_consistent_backup/State.py

Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,5 @@`
`1`	`1`	`from mongodb_consistent_backup.Archive.Tar import Tar`
	`2`	`+from mongodb_consistent_backup.Archive.Zbackup import Zbackup`
`2`	`3`	`from mongodb_consistent_backup.Pipeline import Stage`
`3`	`4`
`4`	`5`