Merge "allow replicator run_once to check specific devices/partitions"

bin/swift-object-replicator

@@ -17,7 +17,15 @@
 from swift.obj.replicator import ObjectReplicator
 from swift.common.utils import parse_options
 from swift.common.daemon import run_daemon
+from optparse import OptionParser
 
 if __name__ == '__main__':
-    conf_file, options = parse_options(once=True)
+    parser = OptionParser("%prog CONFIG [options]")
+    parser.add_option('-d', '--devices',
+                      help='Replicate only given devices. '
+                           'Comma-separated list')
+    parser.add_option('-p', '--partitions',
+                      help='Replicate only given partitions. '
+                           'Comma-separated list')
+    conf_file, options = parse_options(parser=parser, once=True)
     run_daemon(ObjectReplicator, conf_file, **options)

doc/source/admin_guide.rst

@@ -876,6 +876,19 @@ run this command as follows:
 `swift-object-auditor /path/to/object-server/config/file.conf once -z 1000`
 "-z" means to only check for zero-byte files at 1000 files per second.
 
+-----------------
+Object Replicator
+-----------------
+
+At times it is useful to be able to run the object replicator on a specific
+device or partition.  You can run the object-replicator as follows:
+swift-object-replicator /path/to/object-server/config/file.conf once --devices=sda,sdb
+
+This will run the object replicator on only the sda and sdb devices.  You can
+likewise run that command with --partitions.  Both params accept a comma
+separated list of values. If both are specified they will be ANDed together.
+These can only be run in "once" mode.
+
 -------------
 Swift Orphans
 -------------

swift/common/utils.py

@@ -1430,3 +1430,13 @@ def get_valid_utf8_str(str_or_unicode):
         (str_or_unicode, _len) = utf8_encoder(str_or_unicode, 'replace')
     (valid_utf8_str, _len) = utf8_decoder(str_or_unicode, 'replace')
     return valid_utf8_str.encode('utf-8')
+
+
+def list_from_csv(comma_separated_str):
+    """
+    Splits the str given and returns a properly stripped list of the comma
+    separated values.
+    """
+    if comma_separated_str:
+        return [v.strip() for v in comma_separated_str.split(',') if v.strip()]
+    return []

swift/obj/replicator.py

@@ -33,7 +33,7 @@ from eventlet.support.greenlets import GreenletExit
 from swift.common.ring import Ring
 from swift.common.utils import whataremyips, unlink_older_than, lock_path, \
     compute_eta, get_logger, write_pickle, renamer, dump_recon_cache, \
-    rsync_ip, mkdirs, TRUE_VALUES
+    rsync_ip, mkdirs, TRUE_VALUES, list_from_csv
 from swift.common.bufferedhttp import http_connect
 from swift.common.daemon import Daemon
 from swift.common.http import HTTP_OK, HTTP_INSUFFICIENT_STORAGE
@@ -594,7 +594,7 @@ class ObjectReplicator(Daemon):
         self.job_count = len(jobs)
         return jobs
 
-    def replicate(self):
+    def replicate(self, override_devices=[], override_partitions=[]):
         """Run a replication pass"""
         self.start = time.time()
         self.suffix_count = 0
@@ -610,6 +610,11 @@ class ObjectReplicator(Daemon):
             self.run_pool = GreenPool(size=self.concurrency)
             jobs = self.collect_jobs()
             for job in jobs:
+                if override_devices and job['device'] not in override_devices:
+                    continue
+                if override_partitions and \
+                        job['partition'] not in override_partitions:
+                    continue
                 dev_path = join(self.devices_dir, job['device'])
                 if self.mount_check and not os.path.ismount(dev_path):
                     self.logger.warn(_('%s is not mounted'), job['device'])
@@ -635,7 +640,11 @@ class ObjectReplicator(Daemon):
     def run_once(self, *args, **kwargs):
         start = time.time()
         self.logger.info(_("Running object replicator in script mode."))
-        self.replicate()
+        override_devices = list_from_csv(kwargs.get('devices'))
+        override_partitions = list_from_csv(kwargs.get('partitions'))
+        self.replicate(
+            override_devices=override_devices,
+            override_partitions=override_partitions)
         total = (time.time() - start) / 60
         self.logger.info(
             _("Object replication complete. (%.02f minutes)"), total)

test/unit/obj/test_replicator.py

@@ -448,6 +448,21 @@ class TestObjectReplicator(unittest.TestCase):
         self.replicator.replicate()
         self.assertFalse(os.access(part_path, os.F_OK))
 
+    def test_delete_partition_override_params(self):
+        df = DiskFile(self.devices, 'sda', '0', 'a', 'c', 'o', FakeLogger())
+        mkdirs(df.datadir)
+        ohash = hash_path('a', 'c', 'o')
+        data_dir = ohash[-3:]
+        part_path = os.path.join(self.objects, '1')
+        self.assertTrue(os.access(part_path, os.F_OK))
+        self.replicator.replicate(override_devices=['sdb'])
+        self.assertTrue(os.access(part_path, os.F_OK))
+        self.replicator.replicate(override_partitions=['9'])
+        self.assertTrue(os.access(part_path, os.F_OK))
+        self.replicator.replicate(override_devices=['sda'],
+                                  override_partitions=['1'])
+        self.assertFalse(os.access(part_path, os.F_OK))
+
     def test_run_once_recover_from_failure(self):
         replicator = object_replicator.ObjectReplicator(
             dict(swift_dir=self.testdir, devices=self.devices,