Only put the tag-parsing part of the parse_read_pref logic in Common/DB.py, move some specific stuff back to Backup/Mongodump/MongodumpThread.py. Added tag-awareness to Replication/Replset.py so we can score/find members properly

Author: timvaillancourt

mongodb_consistent_backup/Backup/Mongodump/MongodumpThread.py

@@ -9,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, parse_read_pref
+from mongodb_consistent_backup.Common import is_datetime, parse_config_bool, parse_read_pref_tags
 from mongodb_consistent_backup.Oplog import Oplog
 
 
@@ -69,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)
+
     def parse_mongodump_line(self, line):
         try:
             line = line.rstrip()
@@ -149,7 +155,7 @@ def mongodump_cmd(self):
 
         # --readPreference
         if self.is_version_gte("3.2.0"):
-            read_pref = parse_read_pref("secondary", self.read_pref_tags)
+            read_pref = self.parse_read_pref()
             if read_pref:
                 mongodump_flags.append("--readPreference=%s" % json.dumps(read_pref))
         elif self.read_pref_tags:

mongodb_consistent_backup/Common/DB.py

@@ -11,15 +11,13 @@
 from mongodb_consistent_backup.Errors import DBAuthenticationError, DBConnectionError, DBOperationError, Error
 
 
-def parse_read_pref(mode="secondary", tags_str=None):
-    rp = {"mode": mode}
-    if tags_str:
-        rp["tags"] = {}
-        for pair in tags_str.replace(" ", "").split(","):
-            if ":" in pair:
-                key, value = pair.split(":")
-                rp["tags"][key] = str(value)
-    return rp
+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:

mongodb_consistent_backup/Common/__init__.py

@@ -1,5 +1,5 @@
 from Config import Config, parse_config_bool  # NOQA
-from DB import DB, parse_read_pref  # 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/Replication/Replset.py

@@ -4,18 +4,19 @@
 from math import ceil
 from time import mktime
 
-from mongodb_consistent_backup.Common import DB, MongoUri
+from mongodb_consistent_backup.Common import DB, MongoUri, parse_read_pref_tags
 from mongodb_consistent_backup.Errors import Error, OperationError
 
 
 class Replset:
     def __init__(self, config, db):
-        self.config       = config
-        self.db           = db
-        self.max_lag_secs = self.config.replication.max_lag_secs
-        self.min_priority = self.config.replication.min_priority
-        self.max_priority = self.config.replication.max_priority
-        self.hidden_only  = self.config.replication.hidden_only
+        self.config         = config
+        self.db             = db
+        self.read_pref_tags = self.config.replication.read_pref_tags
+        self.max_lag_secs   = self.config.replication.max_lag_secs
+        self.min_priority   = self.config.replication.min_priority
+        self.max_priority   = self.config.replication.max_priority
+        self.hidden_only    = self.config.replication.hidden_only
 
         self.hidden_weight = 0.20
         self.pri0_weight   = 0.10
@@ -139,6 +140,18 @@ def is_member_electable(self, member):
                 return True
         return False
 
+    def has_read_pref_tags(self, member_config):
+        if "tags" not in member_config:
+            raise OperationError("Member config has no 'tags' field!")
+        tags = parse_read_pref_tags(self.read_pref_tags)
+        member_tags = member_config["tags"]
+        for key in tags:
+            if key not in member_tags:
+                return False
+            if member_tags[key] != tags[key]:
+                return False
+        return True
+
     def find_primary(self, force=False, quiet=False):
         if force or not self.primary:
             rs_status = self.get_rs_status(force, quiet)
@@ -170,8 +183,8 @@ def find_secondary(self, force=False, quiet=False):
         self.get_rs_config(force, quiet)
         self.get_mongo_config(force, quiet)
 
-        quorum    = self.get_rs_quorum()
-        rs_name   = rs_status['set']
+        quorum  = self.get_rs_quorum()
+        rs_name = rs_status['set']
 
         if self.secondary and not force:
             return self.secondary
@@ -193,6 +206,12 @@ def find_secondary(self, force=False, quiet=False):
                 score       = self.max_lag_secs * 10
                 score_scale = 100 / score
                 priority    = 0
+
+                if self.read_pref_tags:
+                    if not self.has_read_pref_tags(member_config):
+                        logging.info("Found SECONDARY %s without required read preference tags, skipping" % member_uri)
+                        continue
+
                 if 'hidden' in member_config and member_config['hidden']:
                     score += (score * self.hidden_weight)
                     log_data['hidden'] = True