Support read_preference tags with Common/DB.py

Author: timvaillancourt

conf/mongodb-consistent-backup.example.conf

@@ -4,6 +4,12 @@ production:
   #username: [auth username] (default: none)
   #password: [auth password] (default: none)
   #authdb: [auth database]   (default: admin)
+  #ssl:
+  #  enabled: [true|false]  (default: false)
+  #  insecure: [true|false] (default: false)
+  #  ca_file: [path]
+  #  crl_file: [path]
+  #  client_cert_file: [path]
   log_dir: /var/log/mongodb-consistent-backup
   backup:
     method: mongodump
@@ -14,10 +20,11 @@ production:
   #    compression: [auto|none|gzip] (default: auto - enable gzip if supported)
   #    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/MongodumpThread.py

@@ -8,7 +8,7 @@
 from signal import signal, SIGINT, SIGTERM, SIG_IGN
 from subprocess import Popen, PIPE
 
-from mongodb_consistent_backup.Common import is_datetime
+from mongodb_consistent_backup.Common import is_datetime, parse_config_bool
 from mongodb_consistent_backup.Oplog import Oplog
 
 
@@ -25,10 +25,13 @@ def __init__(self, state, uri, timer, config, base_dir, version, threads=0, dump
         self.threads   = threads
         self.dump_gzip = dump_gzip
 
-        self.user     = self.config.username
-        self.password = self.config.password
-        self.authdb   = self.config.authdb
-        self.binary   = self.config.backup.mongodump.binary
+        self.user                 = 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.binary               = self.config.backup.mongodump.binary
 
         self.timer_name        = "%s-%s" % (self.__class__.__name__, self.uri.replset)
         self.exit_code         = 1
@@ -52,6 +55,12 @@ def close(self, exit_code=None, frame=None):
             self._command.close()
         sys.exit(self.exit_code)
 
+    def do_ssl(self):
+        return parse_config_bool(self.config.ssl.enabled)
+
+    def do_ssl_insecure(self):
+        return parse_config_bool(self.config.ssl.insecure)
+
     def parse_mongodump_line(self, line):
         try:
             line = line.rstrip()
@@ -133,6 +142,16 @@ def mongodump_cmd(self):
             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])
+        if self.do_ssl():
+            mongodump_flags.append("--ssl")
+            if self.ssl_ca_file:
+                mongodump_flags.extend(["--sslCAFile", self.ssl_ca_file])
+            if self.ssl_crl_file:
+                mongodump_flags.extend(["--sslCRLFile", self.ssl_crl_file])
+            if self.client_cert_file:
+                mongodump_flags.extend(["--sslPEMKeyFile", self.ssl_cert_file])
+            if self.do_ssl_insecure():
+                mongodump_flags.extend(["--sslAllowInvalidCertificates", "--sslAllowInvalidHostnames"])
         mongodump_cmd.extend(mongodump_flags)
         return mongodump_cmd
 

mongodb_consistent_backup/Common/Config.py

@@ -8,6 +8,18 @@
 from yconf.util import NestedDict
 
 
+def parse_config_bool(item):
+    try:
+        if isinstance(item, bool):
+            return item
+        elif isinstance(item, str):
+            if item.rstrip().lower() is "true":
+                return True
+        return False
+    except:
+        return False
+
+
 class PrintVersions(Action):
     def __init__(self, option_strings, dest, nargs=0, **kwargs):
         super(PrintVersions, self).__init__(option_strings=option_strings, dest=dest, nargs=nargs, **kwargs)
@@ -54,6 +66,11 @@ def makeParser(self):
         parser.add_argument("-u", "--user", "--username", dest="username", help="MongoDB Authentication Username (for optional auth)", type=str)
         parser.add_argument("-p", "--password", dest="password", help="MongoDB Authentication Password (for optional auth)", type=str)
         parser.add_argument("-a", "--authdb", dest="authdb", help="MongoDB Auth Database (for optional auth - default: admin)", default='admin', type=str)
+        parser.add_argument("--ssl.enabled", dest="ssl.enabled", help="Use SSL secured database connections to MongoDB hosts (default: false)", default=False, action="store_true")
+        parser.add_argument("--ssl.insecure", dest="ssl.insecure", help="Do not validate the SSL certificate and hostname of the server (default: false)", default=False, action="store_true")
+        parser.add_argument("--ssl.ca_file", dest="ssl.ca_file", help="Path to SSL Certificate Authority file in PEM format (default: use OS default CA)", default=None, type=str)
+        parser.add_argument("--ssl.crl_file", dest="ssl.crl_file", help="Path to SSL Certificate Revocation List file in PEM or DER format (for optional cert revocation)", default=None, type=str)
+        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("--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)

mongodb_consistent_backup/Common/DB.py

@@ -4,46 +4,88 @@
 from inspect import currentframe, getframeinfo
 from pymongo import DESCENDING, CursorType, MongoClient
 from pymongo.errors import ConnectionFailure, OperationFailure, ServerSelectionTimeoutError
+from ssl import CERT_REQUIRED, CERT_NONE
 from time import sleep
 
+from mongodb_consistent_backup.Common import parse_config_bool
 from mongodb_consistent_backup.Errors import DBAuthenticationError, DBConnectionError, DBOperationError, Error
 
 
 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.username     = config.username
-        self.password     = config.password
-        self.authdb       = config.authdb
-        self.do_replset   = do_replset
-        self.read_pref    = read_pref
-        self.do_connect   = do_connect
-        self.conn_timeout = conn_timeout
-        self.retries      = retries
+        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
+
+        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.replace(" ", "")
 
         self.replset    = None
         self._conn      = None
         self._is_master = None
+
         self.connect()
         self.auth_if_required()
 
+    def do_ssl(self):
+        return parse_config_bool(self.config.ssl.enabled)
+
+    def do_ssl_insecure(self):
+        return parse_config_bool(self.config.ssl.insecure)
+
+    def client_opts(self):
+        opts = {
+            "connect":                  self.do_connect,
+            "host":                     self.uri.hosts(),
+            "connectTimeoutMS":         self.conn_timeout,
+            "serverSelectionTimeoutMS": self.conn_timeout,
+            "maxPoolSize":              1,
+        }
+        if self.do_replset:
+            self.replset = self.uri.replset
+            opts.update({
+                "replicaSet":         self.replset,
+                "readPreference":     self.read_pref,
+                "readPreferenceTags": self.read_pref_tags,
+                "w":                  "majority"
+            })
+        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,
+                self.ssl_client_cert_file,
+                self.ssl_crl_file,
+                self.do_ssl_insecure()
+            ))
+            opts.update({
+                "ssl":           True,
+                "ssl_ca_certs":  self.ssl_ca_file,
+                "ssl_crlfile":   self.ssl_crl_file,
+                "ssl_certfile":  self.ssl_client_cert_file,
+                "ssl_cert_reqs": CERT_REQUIRED,
+            })
+            if self.do_ssl_insecure():
+                opts["ssl_cert_reqs"] = CERT_NONE
+        return opts
+
     def connect(self):
         try:
-            if self.do_replset:
-                self.replset = self.uri.replset
-            logging.debug("Getting MongoDB connection to %s (replicaSet=%s, readPreference=%s)" % (
-                self.uri, self.replset, self.read_pref
+            logging.debug("Getting MongoDB connection to %s (replicaSet=%s, readPreference=%s, readPreferenceTags=%s, ssl=%s)" % (
+                self.uri,
+                self.replset,
+                self.read_pref,
+                self.read_pref_tags,
+                self.do_ssl(),
             ))
-            conn = MongoClient(
-                connect=self.do_connect,
-                host=self.uri.hosts(),
-                replicaSet=self.replset,
-                readPreference=self.read_pref,
-                connectTimeoutMS=self.conn_timeout,
-                serverSelectionTimeoutMS=self.conn_timeout,
-                maxPoolSize=1,
-                w="majority"
-            )
+            conn = MongoClient(**self.client_opts())
             if self.do_connect:
                 conn['admin'].command({"ping": 1})
         except (ConnectionFailure, OperationFailure, ServerSelectionTimeoutError), e:

mongodb_consistent_backup/Common/__init__.py

@@ -1,4 +1,4 @@
-from Config import Config  # NOQA
+from Config import Config, parse_config_bool  # NOQA
 from DB import DB  # NOQA
 from LocalCommand import LocalCommand  # NOQA
 from Lock import Lock  # NOQA

mongodb_consistent_backup/Replication/__init__.py

@@ -7,6 +7,6 @@ def config(parser):
     parser.add_argument("--replication.min_priority", dest="replication.min_priority", help="Min priority of secondary members for backup (default: 0)", default=0, type=int)
     parser.add_argument("--replication.max_priority", dest="replication.max_priority", help="Max priority of secondary members for backup (default: 1000)", default=1000, type=int)
     parser.add_argument("--replication.hidden_only", dest="replication.hidden_only", help="Only use hidden secondary members for backup (default: false)", default=False, action="store_true")
-    # todo: add tag-specific backup option
-    # parser.add_argument("-replication.use_tag", dest="replication.use_tag", help="Only use secondary members with tag for backup", type=str)
+    parser.add_argument("--replication.read_pref_tags", dest="replication.read_pref_tags", default=None, type=str,
+                        help="Only use members that match replication tags in comma-separated key:value format (default: none)")
     return parser