Merge remote-tracking branch 'upstream/master' into 1.2.0-README-fixes-v0

Author: timvaillancourt

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
 -  Support for Read Preference Tags for selecting specific nodes for backup
 -  Rotation of backups by time or count

conf/mongodb-consistent-backup.example.conf

@@ -17,13 +17,17 @@ 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)
-  #  min_priority: [0-999]     (default: 0)
-  #  max_priority: [2-1000]    (default: 1000)
-  #  hidden_only: [true|false] (default: false)
+  #  max_lag_secs: [1+]                        (default: 10)
+  #  min_priority: [0-999]                     (default: 0)
+  #  max_priority: [2-1000]                    (default: 1000)
+  #  hidden_only: [true|false]                 (default: false)
+  #  read_pref_tags: [key:value,key:value,...] (default: none)
   #sharding:
   #  balancer:
   #    wait_secs: [1+] (default: 300)

mongodb_consistent_backup/Backup/Mongodump/Mongodump.py

@@ -16,11 +16,11 @@
 class Mongodump(Task):
     def __init__(self, manager, config, timer, base_dir, backup_dir, replsets, backup_stop=None, sharding=None):
         super(Mongodump, self).__init__(self.__class__.__name__, manager, config, timer, base_dir, backup_dir)
-        self.compression_method = self.config.backup.mongodump.compression
-        self.binary             = self.config.backup.mongodump.binary
         self.user               = self.config.username
         self.password           = self.config.password
         self.authdb             = self.config.authdb
+        self.compression_method = self.config.backup.mongodump.compression
+        self.binary             = self.config.backup.mongodump.binary
         self.replsets           = replsets
         self.backup_stop        = backup_stop
         self.sharding           = sharding

mongodb_consistent_backup/Backup/Mongodump/MongodumpThread.py

@@ -1,3 +1,4 @@
+import json
 import os
 import logging
 import sys
@@ -8,7 +9,7 @@
 from signal import signal, SIGINT, SIGTERM, SIG_IGN
 from subprocess import Popen, PIPE
 
-from mongodb_consistent_backup.Common import is_datetime, parse_config_bool
+from mongodb_consistent_backup.Common import is_datetime, parse_config_bool, parse_read_pref_tags
 from mongodb_consistent_backup.Oplog import Oplog
 
 
@@ -31,6 +32,7 @@ def __init__(self, state, uri, timer, config, base_dir, version, threads=0, dump
         self.ssl_ca_file          = self.config.ssl.ca_file
         self.ssl_crl_file         = self.config.ssl.crl_file
         self.ssl_client_cert_file = self.config.ssl.client_cert_file
+        self.read_pref_tags       = self.config.replication.read_pref_tags
         self.binary               = self.config.backup.mongodump.binary
 
         self.timer_name        = "%s-%s" % (self.__class__.__name__, self.uri.replset)
@@ -67,6 +69,12 @@ def is_version_gte(self, compare):
                 return True
         return False
 
+    def parse_read_pref(self, mode="secondary"):
+        rp = {"mode": mode}
+        if self.read_pref_tags:
+            rp["tags"] = parse_read_pref_tags(self.read_pref_tags)
+        return json.dumps(rp, separators=(',', ':'))
+
     def parse_mongodump_line(self, line):
         try:
             line = line.rstrip()
@@ -130,37 +138,59 @@ def wait(self):
     def mongodump_cmd(self):
         mongodump_uri   = self.uri.get()
         mongodump_cmd   = [self.binary]
-        mongodump_flags = ["--host", mongodump_uri.host, "--port", str(mongodump_uri.port), "--oplog", "--out", "%s/dump" % self.backup_dir]
+        mongodump_flags = [
+            "--host=%s" % mongodump_uri.host,
+            "--port=%s" % str(mongodump_uri.port),
+            "--oplog",
+            "--out=%s/dump" % self.backup_dir
+        ]
+
+        # --numParallelCollections
         if self.threads > 0:
-            mongodump_flags.extend(["--numParallelCollections=" + str(self.threads)])
+            mongodump_flags.append("--numParallelCollections=%s" % str(self.threads))
 
+        # --gzip
         if self.dump_gzip:
-            mongodump_flags.extend(["--gzip"])
-
-        if self.is_version_gte("3.4.0"):
-            mongodump_flags.extend(["--readPreference=secondary"])
-
+            mongodump_flags.append("--gzip")
+
+        # --readPreference
+        if self.is_version_gte("3.2.0"):
+            read_pref = self.parse_read_pref()
+            if read_pref:
+                mongodump_flags.append("--readPreference=%s" % read_pref)
+        elif self.read_pref_tags:
+            logging.fatal("Mongodump must be >= 3.2.0 to set read preference!")
+            sys.exit(1)
+
+        # --username/--password/--authdb
         if self.authdb and self.authdb != "admin":
             logging.debug("Using database %s for authentication" % self.authdb)
-            mongodump_flags.extend(["--authenticationDatabase", self.authdb])
+            mongodump_flags.append("--authenticationDatabase=%s" % self.authdb)
         if self.user and self.password:
             # >= 3.0.2 supports password input via stdin to mask from ps
             if self.is_version_gte("3.0.2"):
-                mongodump_flags.extend(["-u", self.user, "-p", '""'])
+                mongodump_flags.extend([
+                    "--username=%s" % self.user,
+                    "--password=\"\""
+                ])
                 self.do_stdin_passwd = True
             else:
                 logging.warning("Mongodump is too old to set password securely! Upgrade to mongodump >= 3.0.2 to resolve this")
-                mongodump_flags.extend(["-u", self.user, "-p", self.password])
+                mongodump_flags.extend([
+                    "--username=%s" % self.user,
+                    "--password=%s" % self.password
+                ])
 
+        # --ssl
         if self.do_ssl():
             if self.is_version_gte("2.6.0"):
                 mongodump_flags.append("--ssl")
                 if self.ssl_ca_file:
-                    mongodump_flags.extend(["--sslCAFile", self.ssl_ca_file])
+                    mongodump_flags.append("--sslCAFile=%s" % self.ssl_ca_file)
                 if self.ssl_crl_file:
-                    mongodump_flags.extend(["--sslCRLFile", self.ssl_crl_file])
+                    mongodump_flags.append("--sslCRLFile=%s" % self.ssl_crl_file)
                 if self.client_cert_file:
-                    mongodump_flags.extend(["--sslPEMKeyFile", self.ssl_cert_file])
+                    mongodump_flags.append("--sslPEMKeyFile=%s" % self.ssl_cert_file)
                 if self.do_ssl_insecure():
                     mongodump_flags.extend(["--sslAllowInvalidCertificates", "--sslAllowInvalidHostnames"])
             else:
@@ -182,7 +212,7 @@ def run(self):
             if os.path.isdir(self.dump_dir):
                 rmtree(self.dump_dir)
             os.makedirs(self.dump_dir)
-            logging.debug("Running mongodump cmd: %s" % mongodump_cmd)
+            logging.debug("Running mongodump cmd: %s" % " ".join(mongodump_cmd))
             self._process = Popen(mongodump_cmd, stdin=PIPE, stderr=PIPE)
             self.wait()
             self.exit_code = self._process.returncode

mongodb_consistent_backup/Backup/__init__.py

@@ -2,7 +2,7 @@
 
 
 def config(parser):
-    parser.add_argument("-n", "--backup.name", dest="backup.name", help="Name of the backup set (required)", type=str)
-    parser.add_argument("-l", "--backup.location", dest="backup.location", help="Base path to store the backup data (required)", type=str)
+    parser.add_argument("-n", "--backup.name", dest="backup.name", help="Name of the backup set (default: default)", default='default', type=str)
+    parser.add_argument("-l", "--backup.location", dest="backup.location", help="Base path to store the backup data (default: /var/lib/mongodb-consistent-backup)", default='/var/lib/mongodb-consistent-backup', type=str)
     parser.add_argument("-m", "--backup.method", dest="backup.method", help="Method to be used for backup (default: mongodump)", default='mongodump', choices=['mongodump'])
     return parser

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/Common/DB.py

@@ -11,15 +11,34 @@
 from mongodb_consistent_backup.Errors import DBAuthenticationError, DBConnectionError, DBOperationError, Error
 
 
+def parse_read_pref_tags(tags_str):
+    tags = {}
+    for pair in tags_str.replace(" ", "").split(","):
+        if ":" in pair:
+            key, value = pair.split(":")
+            tags[key] = str(value)
+    return tags
+
+
 class DB:
-    def __init__(self, uri, config, do_replset=False, read_pref='primaryPreferred', do_connect=True, conn_timeout=5000, retries=5):
-        self.uri          = uri
-        self.config       = config
-        self.do_replset   = do_replset
-        self.read_pref    = read_pref
-        self.do_connect   = do_connect
-        self.conn_timeout = conn_timeout
-        self.retries      = retries
+    def __init__(self, uri, config, do_replset=False, read_pref='primaryPreferred', do_rp_tags=False,
+                 do_connect=True, conn_timeout=5000, retries=5):
+        self.uri            = uri
+        self.config         = config
+        self.do_replset     = do_replset
+        self.read_pref      = read_pref
+        self.do_rp_tags     = do_rp_tags
+        self.do_connect     = do_connect
+        self.conn_timeout   = conn_timeout
+        self.retries        = retries
+
+        self.username             = self.config.username
+        self.password             = self.config.password
+        self.authdb               = self.config.authdb
+        self.ssl_ca_file          = self.config.ssl.ca_file
+        self.ssl_crl_file         = self.config.ssl.crl_file
+        self.ssl_client_cert_file = self.config.ssl.client_cert_file
+        self.read_pref_tags       = self.config.replication.read_pref_tags
 
         self.username             = self.config.username
         self.password             = self.config.password
@@ -56,6 +75,14 @@ def client_opts(self):
                 "readPreference": self.read_pref,
                 "w":              "majority"
             })
+            if self.do_rp_tags and self.read_pref_tags:
+                logging.debug("Using read preference mode: %s, tags: %s" % (
+                    self.read_pref,
+                    parse_read_pref_tags(self.read_pref_tags)
+                ))
+                self.read_pref_tags = self.read_pref_tags.replace(" ", "")
+                opts["readPreferenceTags"] = self.read_pref_tags
+
         if self.do_ssl():
             logging.debug("Using SSL-secured mongodb connection (ca_cert=%s, client_cert=%s, crl_file=%s, insecure=%s)" % (
                 self.ssl_ca_file,
@@ -76,10 +103,11 @@ def client_opts(self):
 
     def connect(self):
         try:
-            logging.debug("Getting MongoDB connection to %s (replicaSet=%s, readPreference=%s, ssl=%s)" % (
+            logging.debug("Getting MongoDB connection to %s (replicaSet=%s, readPreference=%s, readPreferenceTags=%s, ssl=%s)" % (
                 self.uri,
                 self.replset,
                 self.read_pref,
+                self.do_rp_tags,
                 self.do_ssl(),
             ))
             conn = MongoClient(**self.client_opts())

mongodb_consistent_backup/Common/__init__.py

@@ -1,5 +1,5 @@
 from Config import Config, parse_config_bool  # NOQA
-from DB import DB  # NOQA
+from DB import DB, parse_read_pref_tags  # NOQA
 from LocalCommand import LocalCommand  # NOQA
 from Lock import Lock  # NOQA
 from MongoUri import MongoUri  # NOQA

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/Oplog/Tailer/TailThread.py

@@ -3,7 +3,7 @@
 
 # noinspection PyPackageRequirements
 from multiprocessing import Process
-from pymongo.errors import AutoReconnect, ConnectionFailure, CursorNotFound, ExceededMaxWaiters, ExecutionTimeout, NetworkTimeout, NotMasterError
+from pymongo.errors import AutoReconnect, ConnectionFailure, CursorNotFound, ExceededMaxWaiters, ExecutionTimeout, NetworkTimeout, NotMasterError, ServerSelectionTimeoutError
 from signal import signal, SIGINT, SIGTERM, SIG_IGN
 from time import sleep, time
 
@@ -88,7 +88,7 @@ def status(self):
 
     def connect(self):
         if not self.db:
-            self.db = DB(self.uri, self.config, True, 'secondary')
+            self.db = DB(self.uri, self.config, True, 'secondary', True)
         return self.db.connection()
 
     def run(self):
@@ -144,21 +144,25 @@ def run(self):
                             continue
                     sleep(1)
                 finally:
-                    self._cursor.close()
+                    if self._cursor:
+                        logging.debug("Stopping oplog cursor on %s" % self.uri)
+                        self._cursor.close()
         except OperationError, e:
             logging.error("Tailer %s encountered error: %s" % (self.uri, e))
             self.exit_code = 1
             self.backup_stop.set()
             raise OperationError(e)
+        except ServerSelectionTimeoutError, e:
+            logging.error("Tailer %s could not connect: %s" % (self.uri, e))
+            self.exit_code = 1
+            self.backup_stop.set()
+            raise OperationError(e)
         except Exception, e:
             logging.error("Tailer %s encountered an unexpected error: %s" % (self.uri, e))
             self.exit_code = 1
             self.backup_stop.set()
             raise e
         finally:
-            if self._cursor:
-                logging.debug("Stopping oplog cursor on %s" % self.uri)
-                self._cursor.close()
             oplog.flush()
             oplog.close()
             self.stopped = True