1.1.0 Release (#188)

* Only apply quorum checking to voting members

* Only apply quorum checking to electable members (not arbiter or priority=0)

* 'make docker' must depend on the bin/mongodb-consistent-backup Makefile-step (#167)

* Fix issue #163, move to OperationError() (#168)

* Remove 'secondary_count' logic, only count electable nodes for new quorum logic (#169)

* 1.0.4: Print more mongodump details (#170)

* Print more details about mongodump

* Version must be set first

* Check if version unknown

* 1.0.4: Ensure Resolver consistent timestamp is not before end of mongodump (#164) (#171)

* Don't return an consistent_end_ts that is earlier than the last_ts of mongodump oplogs
* Some code cleanup and fixed rounding-up logic

* 1.0.4: Support handling of TailThread cursor/connection failures (#174)

Support better handling/logging of TailThread cursor/connection failures
Fsync the tailed oplog to disk on time or doc-writes thresholds.
Pass 'backup_stop' event to threads to signal failure to other child threads.
Allow oplog tailing to be disabled.
Added script to simulate failure of cursor/query.

* Fix disabling of oplog tailer

* Fix disabling of oplog tailer (#175)

* 1.0.4: Mongodump handle error (#176)

* Handle mongodump's "Failed: ..." messages as a proper error

* 1.0.4: Support Google Cloud Storage upload v1 (#178)

* 1.1.0 VERSION (#180)

* Readme gcs (#181)

* Readme gcs 2 (#182)

* 1.1.0 requirements.txt update (#183)

* Update Fabric and boto versions

* 1.1.0 repllag default 10s (#184)

* Raise default repl lag max to 10s, 5s has shown to be too low on many hosts. 3.4 adds a no-op oplog update every 10s so this seems to align nicely with it

* 1.1.0 gcs upload threading (#185)

Author: timvaillancourt

Makefile

@@ -34,7 +34,7 @@ rpm: bin/mongodb-consistent-backup
 	cp -f $(PWD)/conf/mongodb-consistent-backup.example.conf build/rpm/SOURCES/mongodb-consistent-backup.conf
 	rpmbuild -D "_topdir $(PWD)/build/rpm" -D "version $(VERSION)" -bb scripts/$(NAME).spec
 
-docker:
+docker: bin/mongodb-consistent-backup
 	docker build --no-cache --tag $(DOCKER_TAG) --build-arg "RELEASE=$(VERSION)" .
 
 clean:

README.rst

@@ -24,6 +24,7 @@ Features
 -  Block de-duplication and optional AES encryption at rest via `ZBackup <http://zbackup.org/>`__
    archiving method (*optional*)
 -  `AWS S3 <https://aws.amazon.com/s3/>`__ Secure Multipart backup uploads (*optional*)
+-  `Google Cloud Storage <https://cloud.google.com/storage/>`__ Secure backup uploads (*optional*)
 -  `Nagios NSCA <https://sourceforge.net/p/nagios/nsca>`__ push
    notification support (*optional*)
 -  Modular backup, archiving, upload and notification components
@@ -212,7 +213,7 @@ Roadmap
 -  Upload compatibility for ZBackup archive phase *(upload unsupported today)*
 -  Backup retention/rotation *(eg: delete old backups)*
 -  Support more notification methods *(Prometheus, PagerDuty, etc)*
--  Support more upload methods *(Google Cloud Storage, Rsync, etc)*
+-  Support more upload methods *(Rsync, etc)*
 -  Support SSL MongoDB connections
 -  Documentation for running under Docker with persistent volumes
 -  Python unit tests

VERSION

@@ -1 +1 @@
-1.0.3
+1.1.0

conf/mongodb-consistent-backup.example.conf

@@ -14,7 +14,7 @@ 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: 5)
+  #  max_lag_secs: [1+]        (default: 10)
   #  min_priority: [0-999]     (default: 0)
   #  max_priority: [2-1000]    (default: 1000)
   #  hidden_only: [true|false] (default: false)
@@ -24,8 +24,12 @@ production:
   #    ping_secs: [1+] (default: 3)
   #oplog:
   #  compression: [none|gzip] (default: gzip - if gzip is used by backup stage)
+  #  flush:
+  #    max_docs: 100
+  #    max_secs: 1
   #  resolver_threads: [1+]   (default: 2 per CPU)
   #  tailer:
+  #    enabled: true
   #    status_interval: 30
   archive:
     method: tar
@@ -48,6 +52,13 @@ production:
   upload:
     method: none
   #  remove_uploaded: [true|false] (default: false)
+  #  gs:
+  #    project_id: [Google Cloud Project ID]
+  #    access_key: [Google Cloud Storage Access Key]
+  #    secret_key: [Google Cloud Storage Secret Key]
+  #    bucket_name: [Google Cloud Storage Bucket Name]
+  #    bucket_prefix: [prefix] (default: /)
+  #    threads: [1+]           (default: 1 per CPU)
   #  s3:
   #    region: [AWS S3 Region] (default: us-east-1)
   #    access_key: [AWS S3 Access Key]

mongodb_consistent_backup/Backup/Backup.py

@@ -3,7 +3,7 @@
 
 
 class Backup(Stage):
-    def __init__(self, manager, config, timer, base_dir, backup_dir, replsets, sharding=None):
-        super(Backup, self).__init__(self.__class__.__name__, manager, config, timer, base_dir, backup_dir, replsets=replsets, sharding=sharding)
+    def __init__(self, manager, config, timer, base_dir, backup_dir, replsets, backup_stop=None, sharding=None):
+        super(Backup, self).__init__(self.__class__.__name__, manager, config, timer, base_dir, backup_dir, replsets=replsets, backup_stop=backup_stop, sharding=sharding)
         self.task = self.config.backup.method
         self.init()

mongodb_consistent_backup/Backup/Mongodump/Mongodump.py

@@ -2,12 +2,12 @@
 import logging
 import signal
 
-from fabric.api import hide, settings, local
 from math import floor
 from multiprocessing import cpu_count
+from subprocess import check_output
 from time import sleep
 
-from mongodb_consistent_backup.Common import MongoUri
+from mongodb_consistent_backup.Common import MongoUri, config_to_string
 from mongodb_consistent_backup.Errors import Error, OperationError
 from mongodb_consistent_backup.Oplog import OplogState
 from mongodb_consistent_backup.Pipeline import Task
@@ -16,31 +16,46 @@
 
 
 class Mongodump(Task):
-    def __init__(self, manager, config, timer, base_dir, backup_dir, replsets, sharding=None):
+    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.replsets           = replsets
+        self.backup_stop        = backup_stop
         self.sharding           = sharding
 
         self.compression_supported = ['auto', 'none', 'gzip']
         self.version               = 'unknown'
+        self.version_extra         = {}
         self.threads_max           = 16
         self.config_replset        = False
         self.dump_threads          = []
         self.states                = {}
         self._summary              = {}
 
+        self.parse_mongodump_version()
+        self.choose_compression()
+
         if self.config.backup.mongodump.threads and self.config.backup.mongodump.threads > 0:
             self.threads(self.config.backup.mongodump.threads)
 
-        with hide('running', 'warnings'), settings(warn_only=True):
-            self.version = local("%s --version|awk 'NR >1 {exit}; /version/{print $NF}'" % self.binary, capture=True)
-
-        self.choose_compression()
+    def parse_mongodump_version(self):
+        if os.path.isfile(self.binary):
+            output = check_output([self.binary, "--version"])
+            lines  = output.rstrip().split("\n")
+            for line in lines:
+                if "version:" in line:
+                    name, version_num = line.split(" version: ")
+                    if name == 'mongodump':
+                        self.version = version_num
+                        if '-' in version_num:
+                            self.version = version_num.split("-")[0]
+                    self.version_extra[name.lower()] = version_num
+            return self.version, self.version_extra
+        raise OperationError("Could not parse mongodump --version output!")
 
     def choose_compression(self):
         if self.can_gzip():
@@ -76,6 +91,9 @@ def wait(self):
         start_threads = len(self.dump_threads)
         # wait for all threads to finish
         while len(self.dump_threads) > 0:
+            if self.backup_stop and self.backup_stop.is_set():
+                logging.error("Received backup stop event due to error(s), stopping backup!")
+                raise OperationError("Received backup stop event due to error(s)")
             for thread in self.dump_threads:
                 if not thread.is_alive():
                     if thread.exitcode == 0:
@@ -97,7 +115,7 @@ def wait(self):
     def threads(self, threads=None):
         if threads:
             self.thread_count = int(threads)
-        elif not self.thread_count:
+        elif not self.thread_count and self.version is not 'unknown':
             if tuple(self.version.split(".")) >= tuple("3.2.0".split(".")):
                 self.thread_count = 1
                 if self.cpu_count > len(self.replsets):
@@ -120,23 +138,24 @@ def run(self):
                 self.states[shard],
                 mongo_uri,
                 self.timer,
-                self.user,
-                self.password,
-                self.authdb,
+                self.config,
                 self.backup_dir,
-                self.binary,
                 self.version,
                 self.threads(),
-                self.do_gzip(),
-                self.verbose
+                self.do_gzip()
             )
             self.dump_threads.append(thread)
 
         if not len(self.dump_threads) > 0:
             raise OperationError('No backup threads started!')
 
+        options = {
+            'compression':      self.compression(),
+            'threads_per_dump': self.threads()
+        }
+        options.update(self.version_extra)
         logging.info(
-            "Starting backups using mongodump %s (options: compression=%s, threads_per_dump=%i)" % (self.version, self.compression(), self.threads()))
+            "Starting backups using mongodump %s (options: %s)" % (self.version, config_to_string(options)))
         for thread in self.dump_threads:
             thread.start()
         self.wait()
@@ -152,20 +171,17 @@ def run(self):
                     self.states['configsvr'],
                     mongo_uri,
                     self.timer,
-                    self.user,
-                    self.password,
-                    self.authdb,
+                    self.config,
                     self.backup_dir,
-                    self.binary,
                     self.version,
                     self.threads(),
-                    self.do_gzip(),
-                    self.verbose
+                    self.do_gzip()
                 )]
                 self.dump_threads[0].start()
                 self.dump_threads[0].join()
 
         self.completed = True
+        self.stopped   = True
         return self._summary
 
     def close(self):

mongodb_consistent_backup/Backup/Mongodump/MongodumpThread.py

@@ -14,24 +14,25 @@
 
 # noinspection PyStringFormat
 class MongodumpThread(Process):
-    def __init__(self, state, uri, timer, user, password, authdb, base_dir, binary, version,
-                 threads=0, dump_gzip=False, verbose=False):
+    def __init__(self, state, uri, timer, config, base_dir, version, threads=0, dump_gzip=False):
         Process.__init__(self)
         self.state     = state
         self.uri       = uri
         self.timer     = timer
-        self.user      = user
-        self.password  = password
-        self.authdb    = authdb
+        self.config    = config
         self.base_dir  = base_dir
-        self.binary    = binary
         self.version   = version
         self.threads   = threads
         self.dump_gzip = dump_gzip
-        self.verbose   = verbose
+
+        self.user     = self.config.username
+        self.password = self.config.password
+        self.authdb   = self.config.authdb
+        self.binary   = self.config.backup.mongodump.binary
 
         self.timer_name        = "%s-%s" % (self.__class__.__name__, self.uri.replset)
         self.exit_code         = 1
+        self.error_message     = None
         self._command          = None
         self.do_stdin_passwd   = False
         self.stdin_passwd_sent = False
@@ -76,6 +77,17 @@ def handle_password_prompt(self):
             self._process.stdin.flush()
             self.stdin_passwd_sent = True
 
+    def is_failed_line(self, line):
+        if line and line.startswith("Failed: "):
+            return True
+        return False
+
+    def handle_failure(self, line):
+        self.error_message = line.replace("Failed: ", "").capitalize()
+        logging.error("Mongodump error: %s" % self.error_message)
+        self.exit_code = 1
+        self.close()
+
     def wait(self):
         try:
             while self._process.stderr:
@@ -88,6 +100,9 @@ def wait(self):
                             continue
                         elif self.is_password_prompt(read):
                             self.handle_password_prompt()
+                        elif self.is_failed_line(read):
+                            self.handle_failure(read)
+                            break
                         else:
                             logging.info(line)
                 if self._process.poll() != None:

mongodb_consistent_backup/Common/Config.py

@@ -100,7 +100,7 @@ def to_dict(self, data):
             for key in data:
                 value = self.to_dict(data[key])
                 if value and key is not ('merge'):
-                    if key == "password":
+                    if key == "password" or key == "secret_key":
                         value = "******"
                     ret[key] = value
             return ret

mongodb_consistent_backup/Common/DB.py

@@ -1,6 +1,8 @@
 import logging
 
-from pymongo import MongoClient
+from bson.codec_options import CodecOptions
+from inspect import currentframe, getframeinfo
+from pymongo import DESCENDING, CursorType, MongoClient
 from pymongo.errors import ConnectionFailure, OperationFailure, ServerSelectionTimeoutError
 from time import sleep
 
@@ -117,6 +119,26 @@ def replset(self):
             return isMaster['setName']
         return None
 
+    def get_oplog_rs(self):
+        if not self._conn:
+            self.connect()
+        db = self._conn['local']
+        return db.oplog.rs.with_options(codec_options=CodecOptions(unicode_decode_error_handler="ignore"))
+
+    def get_oplog_tail_ts(self):
+        logging.debug("Gathering oldest 'ts' in %s oplog" % self.uri)
+        return self.get_oplog_rs().find_one(sort=[('$natural', DESCENDING)])['ts']
+
+    def get_oplog_cursor_since(self, caller, ts=None):
+        frame   = getframeinfo(currentframe().f_back)
+        comment = "%s:%s;%s:%i" % (caller.__name__, frame.function, frame.filename, frame.lineno)
+        if not ts:
+            ts = self.get_oplog_tail_ts()
+        query = {'ts':{'$gte':ts}}
+        logging.debug("Querying oplog on %s with query: %s" % (self.uri, query))
+        # http://api.mongodb.com/python/current/examples/tailable.html
+        return self.get_oplog_rs().find(query, cursor_type=CursorType.TAILABLE_AWAIT, oplog_replay=True).comment(comment)
+
     def close(self):
         if self._conn:
             logging.debug("Closing connection to: %s" % self.uri)

mongodb_consistent_backup/Common/Util.py

@@ -6,10 +6,10 @@
 
 
 def config_to_string(config):
-    config_vars = ""
+    config_pairs = []
     for key in config:
-        config_vars += "%s=%s, " % (key, config[key])
-    return config_vars[:-1]
+        config_pairs.append("%s=%s" % (key, config[key]))
+    return ", ".join(config_pairs)
 
 def is_datetime(string):
     try: