allow replicator run_once to check specific devices/partitions

Change-Id: If45f77fda269ae6e251579542e70eb71bd11fe2a
This commit is contained in:
David Goetz 2012-09-28 12:24:15 -07:00
parent be8777965f
commit a6c44d2764
5 changed files with 59 additions and 4 deletions

View File

@ -17,7 +17,15 @@
from swift.obj.replicator import ObjectReplicator from swift.obj.replicator import ObjectReplicator
from swift.common.utils import parse_options from swift.common.utils import parse_options
from swift.common.daemon import run_daemon from swift.common.daemon import run_daemon
from optparse import OptionParser
if __name__ == '__main__': 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) run_daemon(ObjectReplicator, conf_file, **options)

View File

@ -876,6 +876,19 @@ run this command as follows:
`swift-object-auditor /path/to/object-server/config/file.conf once -z 1000` `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. "-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 Orphans
------------- -------------

View File

@ -1387,3 +1387,13 @@ def get_valid_utf8_str(str_or_unicode):
(str_or_unicode, _len) = utf8_encoder(str_or_unicode, 'replace') (str_or_unicode, _len) = utf8_encoder(str_or_unicode, 'replace')
(valid_utf8_str, _len) = utf8_decoder(str_or_unicode, 'replace') (valid_utf8_str, _len) = utf8_decoder(str_or_unicode, 'replace')
return valid_utf8_str.encode('utf-8') 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 []

View File

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

View File

@ -448,6 +448,21 @@ class TestObjectReplicator(unittest.TestCase):
self.replicator.replicate() self.replicator.replicate()
self.assertFalse(os.access(part_path, os.F_OK)) 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): def test_run_once_recover_from_failure(self):
replicator = object_replicator.ObjectReplicator( replicator = object_replicator.ObjectReplicator(
dict(swift_dir=self.testdir, devices=self.devices, dict(swift_dir=self.testdir, devices=self.devices,