Updated cnf file, fix mongodump ignoring thread count, fix log rotation

Author: timvaillancourt

conf/mongodb-consistent-backup.example.yml

@@ -1,47 +1,50 @@
 production:
   host: localhost
   port: 27017
-  #username: 
-  #password:
-  #authdb: admin
-  log_dir: /tmp
+  #username: [auth username] (default: none)
+  #password: [auth password] (default: none)
+  #authdb: [auth database]   (default: admin)
+  log_dir: /var/log/mongodb-consistent-backup
   backup:
     method: mongodump
     name: default
-    location: /opt/mongodb/backup
-    mongodump:
-      binary: /usr/bin/mongodump
-    #  compression: gzip (default: true - if supported)
-  replication:
-    max_lag_secs: 5
-    min_priority: 0
-    max_priority: 1000
-    hidden_only: false
-  sharding:
-    balancer:
-      wait_secs: 300
-      ping_secs: 3
-  oplog:
-    #compression: gzip
-    #resolver_threads: 4
+    location: /var/lib/mongodb-consistent-backup
+  #  mongodump:
+  #    binary: [path]           (default: /usr/bin/mongodump)
+  #    compression: [none|gzip] (default: true - if supported)
+  #    threads: [1-16]          (default: auto-generated - shards/cpu)
+  #replication:
+  #  max_lag_secs: [1+]        (default: 5)
+  #  min_priority: [0-999]     (default: 0)
+  #  max_priority: [2-1000]    (default: 1000)
+  #  hidden_only: [true|false] (default: false)
+  #sharding:
+  #  balancer:
+  #    wait_secs: [1+] (default: 300)
+  #    ping_secs: [1+] (default: 3)
+  #oplog:
+  #  compression: [none|gzip] (default: true - if used by backup stage)
+  #  resolver_threads: [1+]   (default: 2 per CPU)
+  #  tailer:
+  #    status_interval: 30
   archive:
     method: tar
-    #tar:
-    #  compression: gzip
-    #  threads: 2
+  #  tar:
+  #    compression: [none|gzip] (default: gzip, none if backup is compressed)
+  #    threads: [1+]            (default: 1 per CPU)
   notify:
     method: none
-    #nsca:
-    #  server: localhost:5667
-    #  password: secret
-    #  check_host: localhost
-    #  check_name: mongodb_consistent_backup
+  #  nsca:
+  #    server: [host:port] (default: localhost:5667)
+  #    password: [password]
+  #    check_host: [nagios host]
+  #    check_name: [nagios check name]
   upload:
     method: none
-    remove_uploaded: false
-    #s3:
-    #  access_key: <AWS S3 Access Key>
-    #  secret_key: <AWS S3 Secret Key>
-    #  bucket_name: <AWS S3 Bucket Name>
-    #  bucket_prefix: /
-    #  threads: 4
+  #  remove_uploaded: [true|false] (default: false)
+  #  s3:
+  #    access_key: [AWS S3 Access Key]
+  #    secret_key: [AWS S3 Secret Key]
+  #    bucket_name: [AWS S3 Bucket Name]
+  #    bucket_prefix: [prefix] (default: /)
+  #    threads: [1+]           (default: 1 per CPU)

mongodb_consistent_backup/Archive/Tar/Tar.py

@@ -24,7 +24,7 @@ def _reduce_method(m):
 class Tar(Task):
     def __init__(self, manager, config, timer, base_dir, backup_dir, **kwargs):
         super(Tar, self).__init__(self.__class__.__name__, manager, config, timer, base_dir, backup_dir, **kwargs)
-	self.compression_method = self.config.archive.tar.compression
+        self.compression_method = self.config.archive.tar.compression
         self.binary             = "tar"
 
         self._pool  = None

mongodb_consistent_backup/Backup/Mongodump/Mongodump.py

@@ -26,12 +26,16 @@ 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       = {}
 
+        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)
 

mongodb_consistent_backup/Backup/Mongodump/__init__.py

@@ -7,4 +7,7 @@ def config(parser):
     parser.add_argument("--backup.mongodump.compression", dest="backup.mongodump.compression",
                         help="Compression method to use on backup (default: gzip)", default="gzip",
                         choices=["none", "gzip"])
+    parser.add_argument("--backup.mongodump.threads", dest="backup.mongodump.threads",
+                        help="Number of threads to use for each mongodump process. There is 1 x mongodump per shard, be careful! (default: shards/CPUs)",
+                        default=0, type=int)
     return parser

mongodb_consistent_backup/Logger.py

@@ -34,6 +34,8 @@ def start(self):
                 self.file_log.setLevel(self.log_level)
                 self.file_log.setFormatter(logging.Formatter(self.log_format))
                 logging.getLogger('').addHandler(self.file_log)
+		print "update_symlink()"
+		self.update_symlink()
         except OSError, e:
             logging.warning("Could not start file log handler, writing to stdout only")
             pass
@@ -42,11 +44,12 @@ def close(self):
         if self.file_log:
             self.file_log.close()
 
-    def compress_last(self):
+    def compress(self):
         gz_log = None
         try:
-            if not os.path.isfile(self.last_log):
+            if not os.path.isfile(self.last_log) or self.last_log == self.backup_log_file:
                 return
+            logging.info("Compressing previous log file")
             gz_file = "%s.gz" % self.last_log
             gz_log  = GzipFile(gz_file, "w+")
             with open(self.last_log) as f:
@@ -57,13 +60,13 @@ def compress_last(self):
             if gz_log:
                 gz_log.close()
 
+    def update_symlink(self):
+        if os.path.islink(self.current_log_file):
+            self.last_log = os.readlink(self.current_log_file)
+            os.remove(self.current_log_file)
+        os.symlink(self.backup_log_file, self.current_log_file)
+
     def rotate(self):
-        if self.do_file_log:
-            if os.path.islink(self.current_log_file):
-                 logging.info("Rotating previous log file")
-                 self.last_log = os.readlink(self.current_log_file)
-                 if self.last_log == self.backup_log_file:
-                     return
-                 os.remove(self.current_log_file)
-                 self.compress_last()
-            os.symlink(self.backup_log_file, self.current_log_file)
+        if self.do_file_log and self.last_log:
+            logging.info("Running rotation of log files")
+            self.compress()

mongodb_consistent_backup/Oplog/Tailer/TailThread.py

@@ -15,7 +15,7 @@
 
 
 class TailThread(Process):
-    def __init__(self, do_stop, uri, config, timer, oplog_file, state, dump_gzip=False, status_secs=15):
+    def __init__(self, do_stop, uri, config, timer, oplog_file, state, dump_gzip=False, status_secs=30):
         Process.__init__(self)
         self.do_stop     = do_stop
         self.uri         = uri

mongodb_consistent_backup/Oplog/Tailer/Tailer.py

@@ -14,14 +14,14 @@
 
 
 class Tailer(Task):
-    def __init__(self, manager, config, timer, base_dir, backup_dir, replsets, status_secs=15):
+    def __init__(self, manager, config, timer, base_dir, backup_dir, replsets):
         super(Tailer, self).__init__(self.__class__.__name__, manager, config, timer, base_dir, backup_dir)
         self.backup_name = self.config.name
         self.user        = self.config.username
         self.password    = self.config.password
         self.authdb      = self.config.authdb
+        self.status_secs = self.config.oplog.tailer.status_interval
         self.replsets    = replsets
-        self.status_secs = status_secs
 
         self.shards   = {}
         self._summary = {}

mongodb_consistent_backup/Oplog/__init__.py

@@ -5,6 +5,7 @@
 
 
 def config(parser):
-    parser.add_argument("--oplog.resolver.threads", dest="oplog.resolver.threads", help="Number of threads to use during resolver step (default: 1-per-CPU)", default=0, type=int)
     parser.add_argument("--oplog.compression", dest="oplog.compression", help="Compression method to use on captured oplog file (default: none)", choices=["none","gzip"], default="none")
+    parser.add_argument("--oplog.resolver.threads", dest="oplog.resolver.threads", help="Number of threads to use during resolver step (default: 1-per-CPU)", default=0, type=int)
+    parser.add_argument("--oplog.tailer.status_interval", dest="oplog.tailer.status_interval", help="Number of seconds to wait between reporting oplog tailer status  (default: 30)", default=30, type=int)
     return parser

mongodb_consistent_backup/Pipeline/Task.py

@@ -23,9 +23,9 @@ 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.thread_count       = 1
         self.timer_name         = self.__class__.__name__
 
         signal(SIGINT, SIG_IGN)

scripts/mongodb_consistent_backup.spec

@@ -51,12 +51,6 @@ install -m 0644 %{SOURCE3} %{buildroot}/usr/share/%{name}/README.rst
 #0 0 * * *	%{run_user}	/usr/bin/mongodb-consistent-backup --config=/etc/mongodb-consistent-backup.yml >/dev/null 2>&1
 EOF
 
-# Change /etc config file to use rpm paths for logs and data
-sed -i \
-	-e s@log_dir:\ /tmp@log_dir:\ %{log_dir}@g \
-	-e s@location:\ /opt/mongodb/backup@location:\ %{data_dir}@g \
-	%{buildroot}%{_sysconfdir}/%{bin_name}.yml
-
 
 %pre
 /usr/bin/getent group %{run_group} >/dev/null 2>&1 || /usr/sbin/groupadd -r %{run_group}