consync: Some more tests and bugfixes.

This commit is contained in:
gholt 2011-06-15 02:01:01 +00:00
parent 599aedae95
commit 6587fd914e
2 changed files with 195 additions and 17 deletions
swift/container
test/unit/container

@ -14,10 +14,12 @@
# limitations under the License.
import os
import time
from time import ctime, time
import random
from struct import unpack_from
from eventlet import sleep
from swift.container import server as container_server
from swift.common import client, direct_client
from swift.common.ring import Ring
@ -49,9 +51,12 @@ class _Iter2FileLikeObject(object):
chunk = self._chunk
self._chunk = ''
return chunk + ''.join(self.iterator)
chunk = ''
chunk = self._chunk
self._chunk = ''
if chunk and len(chunk) <= size:
return chunk
try:
chunk = self.iterator.next()
chunk += self.iterator.next()
except StopIteration:
pass
if len(chunk) <= size:
@ -162,7 +167,7 @@ class ContainerSync(Daemon):
#: Number of containers that had a failure of some type.
self.container_failures = 0
#: Time of last stats report.
self.reported = time.time()
self.reported = time()
swift_dir = conf.get('swift_dir', '/etc/swift')
#: swift.common.ring.Ring for locating containers.
self.container_ring = container_ring or \
@ -177,37 +182,37 @@ class ContainerSync(Daemon):
"""
Runs container sync scans until stopped.
"""
time.sleep(random.random() * self.interval)
sleep(random.random() * self.interval)
while True:
begin = time.time()
begin = time()
all_locs = audit_location_generator(self.devices,
container_server.DATADIR,
mount_check=self.mount_check,
logger=self.logger)
for path, device, partition in all_locs:
self.container_sync(path)
if time.time() - self.reported >= 3600: # once an hour
if time() - self.reported >= 3600: # once an hour
self.report()
elapsed = time.time() - begin
elapsed = time() - begin
if elapsed < self.interval:
time.sleep(self.interval - elapsed)
sleep(self.interval - elapsed)
def run_once(self):
"""
Runs a single container sync scan.
"""
self.logger.info(_('Begin container sync "once" mode'))
begin = time.time()
begin = time()
all_locs = audit_location_generator(self.devices,
container_server.DATADIR,
mount_check=self.mount_check,
logger=self.logger)
for path, device, partition in all_locs:
self.container_sync(path)
if time.time() - self.reported >= 3600: # once an hour
if time() - self.reported >= 3600: # once an hour
self.report()
self.report()
elapsed = time.time() - begin
elapsed = time() - begin
self.logger.info(
_('Container sync "once" mode completed: %.02fs'), elapsed)
@ -219,13 +224,13 @@ class ContainerSync(Daemon):
self.logger.info(
_('Since %(time)s: %(sync)s synced [%(delete)s deletes, %(put)s '
'puts], %(skip)s skipped, %(fail)s failed'),
{'time': time.ctime(self.reported),
{'time': ctime(self.reported),
'sync': self.container_syncs,
'delete': self.container_deletes,
'put': self.container_puts,
'skip': self.container_skips,
'fail': self.container_failures})
self.reported = time.time()
self.reported = time()
self.container_syncs = 0
self.container_deletes = 0
self.container_puts = 0
@ -274,8 +279,8 @@ class ContainerSync(Daemon):
'validate_sync_to_err': err})
self.container_failures += 1
return
stop_at = time.time() + self.container_time
while time.time() < stop_at and sync_point2 < sync_point1:
stop_at = time() + self.container_time
while time() < stop_at and sync_point2 < sync_point1:
rows = broker.get_items_since(sync_point2, 1)
if not rows:
break
@ -295,7 +300,7 @@ class ContainerSync(Daemon):
return
sync_point2 = row['ROWID']
broker.set_x_container_sync_points(None, sync_point2)
while time.time() < stop_at:
while time() < stop_at:
rows = broker.get_items_since(sync_point1, 1)
if not rows:
break

@ -0,0 +1,173 @@
# Copyright (c) 2010-2011 OpenStack, LLC.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import unittest
from swift.container import sync
class FakeRing():
def __init__(self):
self.replica_count = 3
self.devs = [{'ip': '10.0.0.%s' % x, 'port': 1000 + x, 'device': 'sda'}
for x in xrange(3)]
def get_nodes(self, account, container=None, obj=None):
return 1, list(self.devs)
class TestContainerSync(unittest.TestCase):
def test_Iter2FileLikeObject(self):
flo = sync._Iter2FileLikeObject(iter(['123', '4567', '89', '0']))
expect = '1234567890'
got = flo.read(2)
self.assertTrue(len(got) <= 2)
self.assertEquals(got, expect[:len(got)])
expect = expect[len(got):]
got = flo.read(5)
self.assertTrue(len(got) <= 5)
self.assertEquals(got, expect[:len(got)])
expect = expect[len(got):]
self.assertEquals(flo.read(), expect)
self.assertEquals(flo.read(), '')
self.assertEquals(flo.read(2), '')
flo = sync._Iter2FileLikeObject(iter(['123', '4567', '89', '0']))
self.assertEquals(flo.read(), '1234567890')
self.assertEquals(flo.read(), '')
self.assertEquals(flo.read(2), '')
def test_init(self):
cring = FakeRing()
oring = FakeRing()
cs = sync.ContainerSync({}, container_ring=cring, object_ring=oring)
self.assertTrue(cs.container_ring is cring)
self.assertTrue(cs.object_ring is oring)
def test_run_forever(self):
# This runs runs_forever with fakes to succeed for two loops, the first
# causing a report but no interval sleep, the second no report but an
# interval sleep.
time_calls = [0]
sleep_calls = []
audit_location_generator_calls = [0]
def fake_time():
time_calls[0] += 1
returns = [1, # Initialized reported time
1, # Start time
3602, # Is it report time (yes)
3602, # Report time
3602, # Elapsed time for "under interval" (no)
3602, # Start time
3603, # Is it report time (no)
3603, # Elapsed time for "under interval" (yes)
]
if time_calls[0] == len(returns) + 1:
raise Exception('we are now done')
return returns[time_calls[0] - 1]
def fake_sleep(amount):
sleep_calls.append(amount)
def fake_audit_location_generator(*args, **kwargs):
audit_location_generator_calls[0] += 1
# Makes .container_sync() short-circuit because 'path' doesn't end
# with .db
return [('path', 'device', 'partition')]
orig_time = sync.time
orig_sleep = sync.sleep
orig_audit_location_generator = sync.audit_location_generator
try:
sync.time = fake_time
sync.sleep = fake_sleep
sync.audit_location_generator = fake_audit_location_generator
cs = sync.ContainerSync({})
cs.run_forever()
except Exception, err:
if str(err) != 'we are now done':
raise
finally:
sync.time = orig_time
sync.sleep = orig_sleep
sync.audit_location_generator = orig_audit_location_generator
self.assertEquals(time_calls, [9])
self.assertEquals(len(sleep_calls), 2)
self.assertTrue(sleep_calls[0] <= cs.interval)
self.assertTrue(sleep_calls[1] == cs.interval - 1)
self.assertEquals(audit_location_generator_calls, [2])
self.assertEquals(cs.reported, 3602)
def test_run_once(self):
# This runs runs_once with fakes twice, the first causing an interim
# report, the second with no interm report.
time_calls = [0]
audit_location_generator_calls = [0]
def fake_time():
time_calls[0] += 1
returns = [1, # Initialized reported time
1, # Start time
3602, # Is it report time (yes)
3602, # Report time
3602, # End report time
3602, # For elapsed
3602, # Start time
3603, # Is it report time (no)
3604, # End report time
3605, # For elapsed
]
if time_calls[0] == len(returns) + 1:
raise Exception('we are now done')
return returns[time_calls[0] - 1]
def fake_audit_location_generator(*args, **kwargs):
audit_location_generator_calls[0] += 1
# Makes .container_sync() short-circuit because 'path' doesn't end
# with .db
return [('path', 'device', 'partition')]
orig_time = sync.time
orig_audit_location_generator = sync.audit_location_generator
try:
sync.time = fake_time
sync.audit_location_generator = fake_audit_location_generator
cs = sync.ContainerSync({})
cs.run_once()
self.assertEquals(time_calls, [6])
self.assertEquals(audit_location_generator_calls, [1])
self.assertEquals(cs.reported, 3602)
cs.run_once()
except Exception, err:
if str(err) != 'we are now done':
raise
finally:
sync.time = orig_time
sync.audit_location_generator = orig_audit_location_generator
self.assertEquals(time_calls, [10])
self.assertEquals(audit_location_generator_calls, [2])
self.assertEquals(cs.reported, 3604)
if __name__ == '__main__':
unittest.main()