expirer: Only try to delete empty containers
Currently, the object-expirer is a little delete-happy -- after a
container gets processed, an attempt is always made to delete it,
even if we know it will fail with a 409 because either
* we found items that this worker won't process (see the
process/processes config options) or
* we encountered failures processing items and thus didn't
pop all the records off the queue that we *could* process.
The failed DELETE has downsides such as
* adding unnecessary load to the container servers (though
the 409 can hopefully be serviced fairly quickly) and
* evicting container info and listing shard ranges from
memcache, which
* can further load the container-servers (if the expiry queues
have gotten large enough to shard), as all the expirers
doing listings now have to go fetch shard ranges directly
from the DB.
Now, only attempt to delete an expiry container if it appears to be
empty. In any reasonably-sized cluster, we expect dozens to hundreds of
concurrent expirers to be processing the expiry queues. Seeing the
container empty is the most reliable signal that the delete is expected
to succeed.
Change-Id: I8c3f78d1d1850ab501180f884f81f84552617fb7
This commit is contained in:
Tim Burke
parent
8ef530d795
commit
229065d781
@ -262,7 +262,9 @@ class ObjectExpirer(Daemon):
|
||||
task_container, task_object, timestamp_to_delete, and target_path
|
||||
"""
|
||||
for task_account, task_container in task_account_container_list:
|
||||
container_empty = True
|
||||
for o in self.swift.iter_objects(task_account, task_container):
|
||||
container_empty = False
|
||||
if six.PY2:
|
||||
task_object = o['name'].encode('utf8')
|
||||
else:
|
||||
@ -292,6 +294,17 @@ class ObjectExpirer(Daemon):
|
||||
target_account, target_container, target_object]),
|
||||
'delete_timestamp': delete_timestamp,
|
||||
'is_async_delete': is_async}
|
||||
if container_empty:
|
||||
try:
|
||||
self.swift.delete_container(
|
||||
task_account, task_container,
|
||||
acceptable_statuses=(2, HTTP_NOT_FOUND, HTTP_CONFLICT))
|
||||
except (Exception, Timeout) as err:
|
||||
self.logger.exception(
|
||||
'Exception while deleting container %(account)s '
|
||||
'%(container)s %(err)s', {
|
||||
'account': task_account,
|
||||
'container': task_container, 'err': str(err)})
|
||||
|
||||
def run_once(self, *args, **kwargs):
|
||||
"""
|
||||
@ -320,7 +333,6 @@ class ObjectExpirer(Daemon):
|
||||
self.report_objects = 0
|
||||
try:
|
||||
self.logger.debug('Run begin')
|
||||
task_account_container_list_to_delete = list()
|
||||
for task_account, my_index, divisor in \
|
||||
self.iter_task_accounts_to_expire():
|
||||
container_count, obj_count = \
|
||||
@ -342,9 +354,6 @@ class ObjectExpirer(Daemon):
|
||||
[(task_account, task_container) for task_container in
|
||||
self.iter_task_containers_to_expire(task_account)]
|
||||
|
||||
task_account_container_list_to_delete.extend(
|
||||
task_account_container_list)
|
||||
|
||||
# delete_task_iter is a generator to yield a dict of
|
||||
# task_account, task_container, task_object, delete_timestamp,
|
||||
# target_path to handle delete actual object and pop the task
|
||||
@ -359,18 +368,6 @@ class ObjectExpirer(Daemon):
|
||||
pool.spawn_n(self.delete_object, **delete_task)
|
||||
|
||||
pool.waitall()
|
||||
for task_account, task_container in \
|
||||
task_account_container_list_to_delete:
|
||||
try:
|
||||
self.swift.delete_container(
|
||||
task_account, task_container,
|
||||
acceptable_statuses=(2, HTTP_NOT_FOUND, HTTP_CONFLICT))
|
||||
except (Exception, Timeout) as err:
|
||||
self.logger.exception(
|
||||
'Exception while deleting container %(account)s '
|
||||
'%(container)s %(err)s', {
|
||||
'account': task_account,
|
||||
'container': task_container, 'err': str(err)})
|
||||
self.logger.debug('Run end')
|
||||
self.report(final=True)
|
||||
except (Exception, Timeout):
|
||||
|
||||
@ -112,26 +112,31 @@ class TestObjectExpirer(TestCase):
|
||||
self.logger = debug_logger('test-expirer')
|
||||
|
||||
self.ts = make_timestamp_iter()
|
||||
self.empty_time = str(int(time() - 864000))
|
||||
self.past_time = str(int(time() - 86400))
|
||||
self.just_past_time = str(int(time() - 1))
|
||||
self.future_time = str(int(time() + 86400))
|
||||
# Dummy task queue for test
|
||||
self.fake_swift = FakeInternalClient({
|
||||
'.expiring_objects': {
|
||||
# this task container will be checked
|
||||
self.empty_time: [],
|
||||
self.past_time: [
|
||||
# tasks ready for execution
|
||||
self.past_time + '-a0/c0/o0',
|
||||
self.past_time + '-a1/c1/o1',
|
||||
self.past_time + '-a2/c2/o2',
|
||||
self.past_time + '-a3/c3/o3',
|
||||
self.past_time + '-a4/c4/o4',
|
||||
self.past_time + '-a5/c5/o5',
|
||||
self.past_time + '-a6/c6/o6',
|
||||
self.past_time + '-a7/c7/o7',
|
||||
self.past_time + '-a4/c4/o4'],
|
||||
self.just_past_time: [
|
||||
self.just_past_time + '-a5/c5/o5',
|
||||
self.just_past_time + '-a6/c6/o6',
|
||||
self.just_past_time + '-a7/c7/o7',
|
||||
# task objects for unicode test
|
||||
self.past_time + u'-a8/c8/o8\u2661',
|
||||
self.past_time + u'-a9/c9/o9\xf8',
|
||||
# this task will be skipped
|
||||
self.just_past_time + u'-a8/c8/o8\u2661',
|
||||
self.just_past_time + u'-a9/c9/o9\xf8',
|
||||
# this task will be skipped and prevent us from even
|
||||
# *trying* to delete the container
|
||||
self.future_time + '-a10/c10/o10'],
|
||||
# this task container will be skipped
|
||||
self.future_time: [
|
||||
@ -140,14 +145,20 @@ class TestObjectExpirer(TestCase):
|
||||
self.expirer = expirer.ObjectExpirer(self.conf, logger=self.logger,
|
||||
swift=self.fake_swift)
|
||||
|
||||
# target object paths which should be expirerd now
|
||||
self.expired_target_path_list = [
|
||||
# map of times to target object paths which should be expirerd now
|
||||
self.expired_target_paths = {
|
||||
self.past_time: [
|
||||
swob.wsgi_to_str(tgt) for tgt in (
|
||||
'a0/c0/o0', 'a1/c1/o1', 'a2/c2/o2', 'a3/c3/o3', 'a4/c4/o4',
|
||||
)
|
||||
],
|
||||
self.just_past_time: [
|
||||
swob.wsgi_to_str(tgt) for tgt in (
|
||||
'a5/c5/o5', 'a6/c6/o6', 'a7/c7/o7',
|
||||
'a8/c8/o8\xe2\x99\xa1', 'a9/c9/o9\xc3\xb8',
|
||||
)
|
||||
]
|
||||
],
|
||||
}
|
||||
|
||||
def tearDown(self):
|
||||
rmtree(self.rcache)
|
||||
@ -302,7 +313,11 @@ class TestObjectExpirer(TestCase):
|
||||
expected = {
|
||||
self.past_time: [
|
||||
self.past_time + '-' + target_path
|
||||
for target_path in self.expired_target_path_list]}
|
||||
for target_path in self.expired_target_paths[self.past_time]],
|
||||
self.just_past_time: [
|
||||
self.just_past_time + '-' + target_path
|
||||
for target_path
|
||||
in self.expired_target_paths[self.just_past_time]]}
|
||||
self.assertEqual(deleted_objects, expected)
|
||||
|
||||
def test_delete_object(self):
|
||||
@ -569,7 +584,7 @@ class TestObjectExpirer(TestCase):
|
||||
self.assertEqual(
|
||||
self.expirer.logger.get_lines_for_level('info'), [
|
||||
'Pass beginning for task account .expiring_objects; '
|
||||
'2 possible containers; 12 possible objects',
|
||||
'4 possible containers; 12 possible objects',
|
||||
'Pass completed in 0s; 10 objects expired',
|
||||
])
|
||||
|
||||
@ -592,7 +607,10 @@ class TestObjectExpirer(TestCase):
|
||||
x.run_once()
|
||||
self.assertEqual(calls, [([
|
||||
self.make_task(self.past_time, target_path)
|
||||
for target_path in self.expired_target_path_list
|
||||
for target_path in self.expired_target_paths[self.past_time]
|
||||
] + [
|
||||
self.make_task(self.just_past_time, target_path)
|
||||
for target_path in self.expired_target_paths[self.just_past_time]
|
||||
], 2)])
|
||||
|
||||
def test_skip_task_account_without_task_container(self):
|
||||
@ -613,16 +631,32 @@ class TestObjectExpirer(TestCase):
|
||||
my_index = 0
|
||||
divisor = 1
|
||||
|
||||
# empty container gets deleted inline
|
||||
task_account_container_list = [('.expiring_objects', self.empty_time)]
|
||||
with mock.patch.object(self.expirer.swift, 'delete_container') \
|
||||
as mock_delete_container:
|
||||
self.assertEqual(
|
||||
list(self.expirer.iter_task_to_expire(
|
||||
task_account_container_list, my_index, divisor)),
|
||||
[])
|
||||
self.assertEqual(mock_delete_container.mock_calls, [
|
||||
mock.call('.expiring_objects', self.empty_time,
|
||||
acceptable_statuses=(2, 404, 409))])
|
||||
|
||||
task_account_container_list = [('.expiring_objects', self.past_time)]
|
||||
|
||||
expected = [
|
||||
self.make_task(self.past_time, target_path)
|
||||
for target_path in self.expired_target_path_list]
|
||||
for target_path in self.expired_target_paths[self.past_time]]
|
||||
|
||||
with mock.patch.object(self.expirer.swift, 'delete_container') \
|
||||
as mock_delete_container:
|
||||
self.assertEqual(
|
||||
list(self.expirer.iter_task_to_expire(
|
||||
task_account_container_list, my_index, divisor)),
|
||||
expected)
|
||||
# not empty; not deleted
|
||||
self.assertEqual(mock_delete_container.mock_calls, [])
|
||||
|
||||
# the task queue has invalid task object
|
||||
invalid_aco_dict = deepcopy(self.fake_swift.aco_dict)
|
||||
@ -677,7 +711,9 @@ class TestObjectExpirer(TestCase):
|
||||
expected = [
|
||||
self.make_task(self.past_time, target_path,
|
||||
is_async_delete=True)
|
||||
for target_path in self.expired_target_path_list]
|
||||
for target_path in (
|
||||
self.expired_target_paths[self.past_time] +
|
||||
self.expired_target_paths[self.just_past_time])]
|
||||
|
||||
self.assertEqual(
|
||||
list(x.iter_task_to_expire(
|
||||
@ -701,8 +737,10 @@ class TestObjectExpirer(TestCase):
|
||||
self.expirer.run_once()
|
||||
|
||||
# iter_objects is called only for past_time, not future_time
|
||||
self.assertEqual(mock_method.call_args_list,
|
||||
[mock.call('.expiring_objects', self.past_time)])
|
||||
self.assertEqual(mock_method.call_args_list, [
|
||||
mock.call('.expiring_objects', self.empty_time),
|
||||
mock.call('.expiring_objects', self.past_time),
|
||||
mock.call('.expiring_objects', self.just_past_time)])
|
||||
|
||||
def test_object_timestamp_break(self):
|
||||
with mock.patch.object(self.expirer, 'delete_actual_object') \
|
||||
@ -714,7 +752,10 @@ class TestObjectExpirer(TestCase):
|
||||
self.assertEqual(
|
||||
mock_method.call_args_list,
|
||||
[mock.call(target_path, self.past_time, False)
|
||||
for target_path in self.expired_target_path_list])
|
||||
for target_path in self.expired_target_paths[self.past_time]] +
|
||||
[mock.call(target_path, self.just_past_time, False)
|
||||
for target_path
|
||||
in self.expired_target_paths[self.just_past_time]])
|
||||
|
||||
def test_failed_delete_keeps_entry(self):
|
||||
def deliberately_blow_up(actual_obj, timestamp):
|
||||
@ -740,7 +781,11 @@ class TestObjectExpirer(TestCase):
|
||||
mock_method.call_args_list,
|
||||
[mock.call('.expiring_objects', self.past_time,
|
||||
self.past_time + '-' + target_path)
|
||||
for target_path in self.expired_target_path_list])
|
||||
for target_path in self.expired_target_paths[self.past_time]] +
|
||||
[mock.call('.expiring_objects', self.just_past_time,
|
||||
self.just_past_time + '-' + target_path)
|
||||
for target_path
|
||||
in self.expired_target_paths[self.just_past_time]])
|
||||
|
||||
def test_success_gets_counted(self):
|
||||
self.assertEqual(self.expirer.report_objects, 0)
|
||||
@ -776,29 +821,37 @@ class TestObjectExpirer(TestCase):
|
||||
raise Exception('failed to delete container')
|
||||
|
||||
def fail_delete_actual_object(actual_obj, timestamp, is_async_delete):
|
||||
if timestamp == self.just_past_time:
|
||||
raise Exception('failed to delete actual object')
|
||||
|
||||
with mock.patch.object(self.fake_swift, 'delete_container',
|
||||
fail_delete_container), \
|
||||
mock.patch.object(self.expirer, 'delete_actual_object',
|
||||
fail_delete_actual_object):
|
||||
fail_delete_actual_object), \
|
||||
mock.patch.object(self.expirer, 'pop_queue') as mock_pop:
|
||||
self.expirer.run_once()
|
||||
|
||||
error_lines = self.expirer.logger.get_lines_for_level('error')
|
||||
|
||||
self.assertEqual(error_lines, [
|
||||
'Exception while deleting container .expiring_objects %s failed '
|
||||
'to delete container: ' % self.empty_time
|
||||
] + [
|
||||
'Exception while deleting object %s %s %s '
|
||||
'failed to delete actual object: ' % (
|
||||
'.expiring_objects', self.past_time,
|
||||
self.past_time + '-' + target_path)
|
||||
for target_path in self.expired_target_path_list] + [
|
||||
'Exception while deleting container %s %s '
|
||||
'failed to delete container: ' % (
|
||||
'.expiring_objects', self.past_time)])
|
||||
'.expiring_objects', self.just_past_time,
|
||||
self.just_past_time + '-' + target_path)
|
||||
for target_path in self.expired_target_paths[self.just_past_time]
|
||||
])
|
||||
self.assertEqual(self.expirer.logger.get_lines_for_level('info'), [
|
||||
'Pass beginning for task account .expiring_objects; '
|
||||
'2 possible containers; 12 possible objects',
|
||||
'Pass completed in 0s; 0 objects expired',
|
||||
'4 possible containers; 12 possible objects',
|
||||
'Pass completed in 0s; 5 objects expired',
|
||||
])
|
||||
self.assertEqual(mock_pop.mock_calls, [
|
||||
mock.call('.expiring_objects', self.past_time,
|
||||
self.past_time + '-' + target_path)
|
||||
for target_path in self.expired_target_paths[self.past_time]
|
||||
])
|
||||
|
||||
def test_run_forever_initial_sleep_random(self):
|
||||
|
||||
Loading…
Reference in New Issue
Block a user