From 3fc4d6f91d48a7b68f108e401a5ee8e2fcdc0100 Mon Sep 17 00:00:00 2001 From: Clay Gerrard Date: Mon, 28 Apr 2014 20:03:48 -0700 Subject: [PATCH] Add container-reconciler daemon This daemon will take objects that are in the wrong storage policy and move them to the right ones, or delete requests that went to the wrong storage policy and apply them to the right ones. It operates on a queue similar to the object-expirer's queue. Discovering that the object is in the wrong policy will be done in subsequent commits by the container replicator; this is the daemon that handles them once they happen. Like the object expirer, you only need to run one of these per cluster see etc/container-reconciler.conf. DocImpact Implements: blueprint storage-policies Change-Id: I5ea62eb77ddcbc7cfebf903429f2ee4c098771c9 --- bin/swift-container-reconciler | 21 + bin/swift-reconciler-enqueue | 74 ++ etc/container-reconciler.conf-sample | 52 + setup.cfg | 2 + swift/common/manager.py | 5 +- swift/common/utils.py | 13 + swift/container/reconciler.py | 749 +++++++++++ test/unit/__init__.py | 5 +- test/unit/common/middleware/helpers.py | 41 +- test/unit/common/test_utils.py | 10 + test/unit/container/test_reconciler.py | 1697 ++++++++++++++++++++++++ 11 files changed, 2658 insertions(+), 11 deletions(-) create mode 100755 bin/swift-container-reconciler create mode 100755 bin/swift-reconciler-enqueue create mode 100644 etc/container-reconciler.conf-sample create mode 100644 swift/container/reconciler.py create mode 100644 test/unit/container/test_reconciler.py diff --git a/bin/swift-container-reconciler b/bin/swift-container-reconciler new file mode 100755 index 0000000000..29f35d8a82 --- /dev/null +++ b/bin/swift-container-reconciler @@ -0,0 +1,21 @@ +#!/usr/bin/env python +# 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. + +from swift.container.reconciler import ContainerReconciler +from swift.common.utils import parse_options +from swift.common.daemon import run_daemon + +if __name__ == '__main__': + conf_file, options = parse_options(once=True) + run_daemon(ContainerReconciler, conf_file, **options) diff --git a/bin/swift-reconciler-enqueue b/bin/swift-reconciler-enqueue new file mode 100755 index 0000000000..662eb393fd --- /dev/null +++ b/bin/swift-reconciler-enqueue @@ -0,0 +1,74 @@ +#!/usr/bin/env python +# 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 sys +from optparse import OptionParser + +import eventlet.debug +eventlet.debug.hub_exceptions(True) + +from swift.common.ring import Ring +from swift.common.utils import split_path +from swift.common.storage_policy import POLICIES + +from swift.container.reconciler import add_to_reconciler_queue +""" +This tool is primarly for debugging and development but can be used an example +of how an operator could enqueue objects manually if a problem is discovered - +might be particularlly useful if you need to hack a fix into the reconciler +and re-run it. +""" + +USAGE = """ +%prog [options] + +This script enqueues an object to be evaluated by the reconciler. + +Arguments: +policy_index: the policy the object is currently stored in. + /a/c/o: the full path of the object - utf-8 + timestamp: the timestamp of the datafile/tombstone. + +""".strip() + +parser = OptionParser(USAGE) +parser.add_option('-X', '--op', default='PUT', choices=('PUT', 'DELETE'), + help='the method of the misplaced operation') +parser.add_option('-f', '--force', action='store_true', + help='force an object to be re-enqueued') + + +def main(): + options, args = parser.parse_args() + try: + policy_index, path, timestamp = args + except ValueError: + sys.exit(parser.print_help()) + container_ring = Ring('/etc/swift/container.ring.gz') + policy = POLICIES.get_by_index(policy_index) + if not policy: + return 'ERROR: invalid storage policy index: %s' % policy + try: + account, container, obj = split_path(path, 3, 3, True) + except ValueError as e: + return 'ERROR: %s' % e + container_name = add_to_reconciler_queue( + container_ring, account, container, obj, + policy.idx, timestamp, options.op, force=options.force) + if not container_name: + return 'ERROR: unable to enqueue!' + print container_name + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/etc/container-reconciler.conf-sample b/etc/container-reconciler.conf-sample new file mode 100644 index 0000000000..0a3de0220d --- /dev/null +++ b/etc/container-reconciler.conf-sample @@ -0,0 +1,52 @@ +[DEFAULT] +# swift_dir = /etc/swift +# user = swift +# You can specify default log routing here if you want: +# log_name = swift +# log_facility = LOG_LOCAL0 +# log_level = INFO +# log_address = /dev/log +# +# comma separated list of functions to call to setup custom log handlers. +# functions get passed: conf, name, log_to_console, log_route, fmt, logger, +# adapted_logger +# log_custom_handlers = +# +# If set, log_udp_host will override log_address +# log_udp_host = +# log_udp_port = 514 +# +# You can enable StatsD logging here: +# log_statsd_host = localhost +# log_statsd_port = 8125 +# log_statsd_default_sample_rate = 1.0 +# log_statsd_sample_rate_factor = 1.0 +# log_statsd_metric_prefix = + +[container-reconciler] +# The reconciler will re-attempt reconciliation if the source object is not +# available up to reclaim_age seconds before it gives up and deletes the entry +# in the queue. +# reclaim_age = 604800 +# The cycle time of the daemon +# interval = 300 +# Server errors from requests will be retried by default +# request_tries = 3 + +[pipeline:main] +pipeline = catch_errors proxy-logging cache proxy-server + +[app:proxy-server] +use = egg:swift#proxy +# See proxy-server.conf-sample for options + +[filter:cache] +use = egg:swift#memcache +# See proxy-server.conf-sample for options + +[filter:proxy-logging] +use = egg:swift#proxy_logging + +[filter:catch_errors] +use = egg:swift#catch_errors +# See proxy-server.conf-sample for options diff --git a/setup.cfg b/setup.cfg index 6239e68ed5..0c35d53f04 100644 --- a/setup.cfg +++ b/setup.cfg @@ -39,6 +39,8 @@ scripts = bin/swift-container-server bin/swift-container-sync bin/swift-container-updater + bin/swift-container-reconciler + bin/swift-reconciler-enqueue bin/swift-dispersion-populate bin/swift-dispersion-report bin/swift-drive-audit diff --git a/swift/common/manager.py b/swift/common/manager.py index 103b5c2979..583912b3c2 100644 --- a/swift/common/manager.py +++ b/swift/common/manager.py @@ -30,7 +30,8 @@ RUN_DIR = '/var/run/swift' # auth-server has been removed from ALL_SERVERS, start it explicitly ALL_SERVERS = ['account-auditor', 'account-server', 'container-auditor', - 'container-replicator', 'container-server', 'container-sync', + 'container-replicator', 'container-reconciler', + 'container-server', 'container-sync', 'container-updater', 'object-auditor', 'object-server', 'object-expirer', 'object-replicator', 'object-updater', 'proxy-server', 'account-replicator', 'account-reaper'] @@ -41,7 +42,7 @@ GRACEFUL_SHUTDOWN_SERVERS = MAIN_SERVERS + ['auth-server'] START_ONCE_SERVERS = REST_SERVERS # These are servers that match a type (account-*, container-*, object-*) but # don't use that type-server.conf file and instead use their own. -STANDALONE_SERVERS = ['object-expirer'] +STANDALONE_SERVERS = ['object-expirer', 'container-reconciler'] KILL_WAIT = 15 # seconds to wait for servers to die (by default) WARNING_WAIT = 3 # seconds to wait after message that may just be a warning diff --git a/swift/common/utils.py b/swift/common/utils.py index 02e9cfe572..1a5b592269 100644 --- a/swift/common/utils.py +++ b/swift/common/utils.py @@ -49,6 +49,7 @@ import glob from urlparse import urlparse as stdlib_urlparse, ParseResult import itertools import stat +import datetime import eventlet import eventlet.semaphore @@ -576,6 +577,18 @@ def normalize_timestamp(timestamp): return "%016.05f" % (float(timestamp)) +def last_modified_date_to_timestamp(last_modified_date_str): + """ + Convert a last modified date (liked you'd get from a container listing, + e.g. 2014-02-28T23:22:36.698390) to a float. + """ + return float( + datetime.datetime.strptime( + last_modified_date_str, '%Y-%m-%dT%H:%M:%S.%f' + ).strftime('%s.%f') + ) + + def normalize_delete_at_timestamp(timestamp): """ Format a timestamp (string or numeric) into a standardized diff --git a/swift/container/reconciler.py b/swift/container/reconciler.py new file mode 100644 index 0000000000..64aa7ba373 --- /dev/null +++ b/swift/container/reconciler.py @@ -0,0 +1,749 @@ +# 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 time +from collections import defaultdict +import socket +import itertools +import logging + +from eventlet import GreenPile, GreenPool, Timeout + +from swift.common import constraints +from swift.common.daemon import Daemon +from swift.common.direct_client import ( + direct_head_container, direct_delete_container_object, + direct_put_container_object, ClientException) +from swift.common.internal_client import InternalClient, UnexpectedResponse +from swift.common.storage_policy import POLICY_INDEX +from swift.common.utils import get_logger, split_path, quorum_size, \ + FileLikeIter, normalize_timestamp, last_modified_date_to_timestamp + + +MISPLACED_OBJECTS_ACCOUNT = '.misplaced_objects' +MISPLACED_OBJECTS_CONTAINER_DIVISOR = 3600 # 1 hour + + +def cmp_policy_info(info, remote_info): + """ + You have to squint to see it, but the general strategy is just: + + if either has been recreated: + return the newest (of the recreated) + else + return the oldest + + I tried cleaning it up for awhile, but settled on just writing a bunch of + tests instead. Once you get an intuitive sense for the nuance here you + can try and see there's a better way to spell the boolean logic but it all + ends up looking sorta hairy. + + :returns: -1 if info is correct, 1 if remote_info is better + """ + def is_deleted(info): + return (info['delete_timestamp'] > info['put_timestamp'] and + info.get('count', info.get('object_count', 0)) == 0) + + deleted = is_deleted(info) + remote_deleted = is_deleted(remote_info) + if any([deleted, remote_deleted]): + if not deleted: + return -1 + elif not remote_deleted: + return 1 + return cmp(remote_info['status_changed_at'], + info['status_changed_at']) + + def has_been_recreated(info): + return (info['put_timestamp'] > info['delete_timestamp'] > + normalize_timestamp(0)) + + remote_recreated = has_been_recreated(remote_info) + recreated = has_been_recreated(info) + if any([remote_recreated, recreated]): + if not recreated: + return 1 + elif not remote_recreated: + return -1 + return cmp(remote_info['status_changed_at'], + info['status_changed_at']) + return cmp(info['status_changed_at'], remote_info['status_changed_at']) + + +def incorrect_policy_index(info, remote_info): + """ + Compare remote_info to info and decide if the remote storage policy index + should be used instead of ours. + """ + if 'storage_policy_index' not in remote_info: + return False + if remote_info['storage_policy_index'] == \ + info['storage_policy_index']: + return False + + return info['storage_policy_index'] != sorted( + [info, remote_info], cmp=cmp_policy_info)[0]['storage_policy_index'] + + +def translate_container_headers_to_info(headers): + default_timestamp = normalize_timestamp(0) + return { + 'storage_policy_index': int(headers[POLICY_INDEX]), + 'put_timestamp': headers.get('x-backend-put-timestamp', + default_timestamp), + 'delete_timestamp': headers.get('x-backend-delete-timestamp', + default_timestamp), + 'status_changed_at': headers.get('x-backend-status-changed-at', + default_timestamp), + } + + +def best_policy_index(headers): + container_info = map(translate_container_headers_to_info, headers) + container_info.sort(cmp=cmp_policy_info) + return container_info[0]['storage_policy_index'] + + +def get_reconciler_container_name(obj_timestamp): + return str(int(float(obj_timestamp)) // + MISPLACED_OBJECTS_CONTAINER_DIVISOR * + MISPLACED_OBJECTS_CONTAINER_DIVISOR) + + +def get_reconciler_obj_name(policy_index, account, container, obj): + return "%(policy_index)d:/%(acc)s/%(con)s/%(obj)s" % { + 'policy_index': policy_index, 'acc': account, + 'con': container, 'obj': obj} + + +def get_reconciler_content_type(op): + try: + return { + 'put': 'application/x-put', + 'delete': 'application/x-delete', + }[op.lower()] + except KeyError: + raise ValueError('invalid operation type %r' % op) + + +def get_row_to_q_entry_translater(broker): + account = broker.account + container = broker.container + op_type = { + 0: get_reconciler_content_type('put'), + 1: get_reconciler_content_type('delete'), + } + + def translater(obj_info): + name = get_reconciler_obj_name(obj_info['storage_policy_index'], + account, container, + obj_info['name']) + return { + 'name': name, + 'deleted': 0, + 'created_at': obj_info['created_at'], + 'etag': obj_info['created_at'], + 'content_type': op_type[obj_info['deleted']], + 'size': 0, + } + return translater + + +def add_to_reconciler_queue(container_ring, account, container, obj, + obj_policy_index, obj_timestamp, op, + force=False, conn_timeout=5, response_timeout=15): + """ + Add an object to the container reconciler's queue. This will cause the + container reconciler to move it from its current storage policy index to + the correct storage policy index. + + :param container_ring: container ring + :param account: the misplaced object's account + :param container: the misplaced object's container + :param obj: the misplaced object + :param obj_policy_index: the policy index where the misplaced object + currently is + :param obj_timestamp: the misplaced object's X-Timestamp. We need this to + ensure that the reconciler doesn't overwrite a newer + object with an older one. + :param op: the method of the operation (DELETE or PUT) + :param force: over-write queue entries newer than obj_timestamp + :param conn_timeout: max time to wait for connection to container server + :param response_timeout: max time to wait for response from container + server + + :returns: .misplaced_object container name, False on failure. "Success" + means a quorum of containers got the update. + """ + container_name = get_reconciler_container_name(obj_timestamp) + object_name = get_reconciler_obj_name(obj_policy_index, account, + container, obj) + if force: + # this allows an operator to re-enqueue an object that has + # already been popped from the queue to be reprocessed, but + # could potentially prevent out of order updates from making it + # into the queue + x_timestamp = normalize_timestamp(time.time()) + else: + x_timestamp = obj_timestamp + q_op_type = get_reconciler_content_type(op) + headers = { + 'X-Size': 0, + 'X-Etag': obj_timestamp, + 'X-Timestamp': x_timestamp, + 'X-Content-Type': q_op_type, + } + + def _check_success(*args, **kwargs): + try: + direct_put_container_object(*args, **kwargs) + return 1 + except (ClientException, Timeout, socket.error): + return 0 + + pile = GreenPile() + part, nodes = container_ring.get_nodes(MISPLACED_OBJECTS_ACCOUNT, + container_name) + for node in nodes: + pile.spawn(_check_success, node, part, MISPLACED_OBJECTS_ACCOUNT, + container_name, object_name, headers=headers, + conn_timeout=conn_timeout, + response_timeout=response_timeout) + + successes = sum(pile) + if successes >= quorum_size(len(nodes)): + return container_name + else: + return False + + +def slightly_later_timestamp(ts, offset=1): + # I'm guessing to avoid rounding errors Swift uses a 10-microsecond + # resolution instead of Python's 1-microsecond resolution. + offset *= 0.00001 + return normalize_timestamp(float(ts) + offset) + + +def parse_raw_obj(obj_info): + """ + Translate a reconciler container listing entry to a dictionary + containing the parts of the misplaced object queue entry. + + :param obj_info: an entry in an a container listing with the + required keys: name, content_type, and hash + + :returns: a queue entry dict with the keys: q_policy_index, account, + container, obj, q_op, q_ts, q_record, and path + """ + raw_obj_name = obj_info['name'].encode('utf-8') + + policy_index, obj_name = raw_obj_name.split(':', 1) + q_policy_index = int(policy_index) + account, container, obj = split_path(obj_name, 3, 3, rest_with_last=True) + try: + q_op = { + 'application/x-put': 'PUT', + 'application/x-delete': 'DELETE', + }[obj_info['content_type']] + except KeyError: + raise ValueError('invalid operation type %r' % + obj_info.get('content_type', None)) + return { + 'q_policy_index': q_policy_index, + 'account': account, + 'container': container, + 'obj': obj, + 'q_op': q_op, + 'q_ts': float(obj_info['hash']), + 'q_record': last_modified_date_to_timestamp( + obj_info['last_modified']), + 'path': '/%s/%s/%s' % (account, container, obj) + } + + +def direct_get_container_policy_index(container_ring, account_name, + container_name): + """ + Talk directly to the primary container servers to figure out the storage + policy index for a given container. + + :param container_ring: ring in which to look up the container locations + :param account_name: name of the container's account + :param container_name: name of the container + :returns: storage policy index, or None if it couldn't get a quorum + """ + def _eat_client_exception(*args): + try: + return direct_head_container(*args) + except ClientException as err: + if err.http_status == 404: + return err.http_headers + except (Timeout, socket.error): + pass + + pile = GreenPile() + part, nodes = container_ring.get_nodes(account_name, container_name) + for node in nodes: + pile.spawn(_eat_client_exception, node, part, account_name, + container_name) + + headers = [x for x in pile if x is not None] + if len(headers) < quorum_size(len(nodes)): + return + return best_policy_index(headers) + + +def direct_delete_container_entry(container_ring, account_name, container_name, + object_name, headers=None): + """ + Talk directly to the primary container servers to delete a particular + object listing. Does not talk to object servers; use this only when a + container entry does not actually have a corresponding object. + """ + pool = GreenPool() + part, nodes = container_ring.get_nodes(account_name, container_name) + for node in nodes: + pool.spawn_n(direct_delete_container_object, node, part, account_name, + container_name, object_name, headers=headers) + + # This either worked or it didn't; if it didn't, we'll retry on the next + # reconciler loop when we see the queue entry again. + pool.waitall() + + +class ContainerReconciler(Daemon): + """ + Move objects that are in the wrong storage policy. + """ + + def __init__(self, conf): + self.conf = conf + self.reclaim_age = int(conf.get('reclaim_age', 86400 * 7)) + self.interval = int(conf.get('interval', 30)) + conf_path = conf.get('__file__') or \ + '/etc/swift/container-reconciler.conf' + self.logger = get_logger(conf, log_route='container-reconciler') + request_tries = int(conf.get('request_tries') or 3) + self.swift = InternalClient(conf_path, + 'Swift Container Reconciler', + request_tries) + self.stats = defaultdict(int) + self.last_stat_time = time.time() + + def stats_log(self, metric, msg, *args, **kwargs): + """ + Update stats tracking for metric and emit log message. + """ + level = kwargs.pop('level', logging.DEBUG) + log_message = '%s: ' % metric + msg + self.logger.log(level, log_message, *args, **kwargs) + self.stats[metric] += 1 + + def log_stats(self, force=False): + """ + Dump stats to logger, noop when stats have been already been + logged in the last minute. + """ + now = time.time() + should_log = force or (now - self.last_stat_time > 60) + if should_log: + self.last_stat_time = now + self.logger.info('Reconciler Stats: %r', dict(**self.stats)) + + def pop_queue(self, container, obj, q_ts, q_record): + """ + Issue a delete object request to the container for the misplaced + object queue entry. + + :param container: the misplaced objects container + :param q_ts: the timestamp of the misplaced object + :param q_record: the timestamp of the queue entry + + N.B. q_ts will normally be the same time as q_record except when + an object was manually re-enqued. + """ + q_path = '/%s/%s/%s' % (MISPLACED_OBJECTS_ACCOUNT, container, obj) + x_timestamp = slightly_later_timestamp(q_record) + self.stats_log('pop_queue', 'remove %r (%f) from the queue (%s)', + q_path, q_ts, x_timestamp) + headers = {'X-Timestamp': x_timestamp} + direct_delete_container_entry( + self.swift.container_ring, MISPLACED_OBJECTS_ACCOUNT, + container, obj, headers=headers) + + def throw_tombstones(self, account, container, obj, timestamp, + policy_index, path): + """ + Issue a delete object request to the given storage_policy. + + :param account: the account name + :param container: the container name + :param account: the object name + :param timestamp: the timestamp of the object to delete + :param policy_index: the policy index to direct the request + :param path: the path to be used for logging + """ + x_timestamp = slightly_later_timestamp(timestamp) + self.stats_log('cleanup_attempt', '%r (%f) from policy_index ' + '%s (%s) will be deleted', + path, timestamp, policy_index, x_timestamp) + headers = { + 'X-Timestamp': x_timestamp, + 'X-Backend-Storage-Policy-Index': policy_index, + } + success = False + try: + self.swift.delete_object(account, container, obj, + acceptable_statuses=(2, 4), + headers=headers) + except UnexpectedResponse as err: + self.stats_log('cleanup_failed', '%r (%f) was not cleaned up ' + 'in storage_policy %s (%s)', path, timestamp, + policy_index, err) + else: + success = True + self.stats_log('cleanup_success', '%r (%f) was successfully ' + 'removed from policy_index %s', path, timestamp, + policy_index) + return success + + def _reconcile_object(self, account, container, obj, q_policy_index, q_ts, + q_op, path, **kwargs): + """ + Perform object reconciliation. + + :param account: the account name of the misplaced object + :param container: the container name of the misplaced object + :param obj: the object name + :param q_policy_index: the policy index of the source indicated by the + queue entry. + :param q_ts: a float, the timestamp of the misplaced object + :param q_op: the operation of the misplaced request + :param path: the full path of the misplaced object for logging + + :returns: True to indicate the request is fully processed + successfully, otherwise False. + """ + container_policy_index = direct_get_container_policy_index( + self.swift.container_ring, account, container) + if container_policy_index is None: + self.stats_log('unavailable_container', '%r (%f) unable to ' + 'determine the destination policy_index', + path, q_ts) + return False + if container_policy_index == q_policy_index: + self.stats_log('noop_object', '%r (%f) container policy_index ' + '%s matches queue policy index %s', path, q_ts, + container_policy_index, q_policy_index) + return True + + # check if object exists in the destination already + self.logger.debug('checking for %r (%f) in destination ' + 'policy_index %s', path, q_ts, + container_policy_index) + headers = { + 'X-Backend-Storage-Policy-Index': container_policy_index} + dest_obj = self.swift.get_object_metadata(account, container, obj, + headers=headers, + acceptable_statuses=(2, 4)) + dest_ts = float( + dest_obj.get('x-timestamp', + dest_obj.get('x-backend-timestamp', + '0.0') + ) + ) + if dest_ts >= q_ts: + self.stats_log('found_object', '%r (%f) in policy_index %s ' + 'is newer than queue (%f)', path, dest_ts, + container_policy_index, q_ts) + return self.throw_tombstones(account, container, obj, q_ts, + q_policy_index, path) + + # object is misplaced + self.stats_log('misplaced_object', '%r (%f) in policy_index %s ' + 'should be in policy_index %s', path, q_ts, + q_policy_index, container_policy_index) + + # fetch object from the source location + self.logger.debug('fetching %r (%f) from storage policy %s', path, + q_ts, q_policy_index) + headers = { + 'X-Backend-Storage-Policy-Index': q_policy_index} + try: + source_obj_status, source_obj_info, source_obj_iter = \ + self.swift.get_object(account, container, obj, + headers=headers, + acceptable_statuses=(2, 4)) + except UnexpectedResponse as err: + source_obj_status = err.resp.status_int + source_obj_info = {} + source_obj_iter = None + + source_ts = float(source_obj_info.get("X-Timestamp", 0)) + if source_obj_status == 404 and q_op == 'DELETE': + return self.ensure_tombstone_in_right_location( + q_policy_index, account, container, obj, q_ts, path, + container_policy_index, source_ts) + else: + return self.ensure_object_in_right_location( + q_policy_index, account, container, obj, q_ts, path, + container_policy_index, source_ts, source_obj_status, + source_obj_info, source_obj_iter) + + def ensure_object_in_right_location(self, q_policy_index, account, + container, obj, q_ts, path, + container_policy_index, source_ts, + source_obj_status, source_obj_info, + source_obj_iter, **kwargs): + """ + Validate source object will satisfy the misplaced object queue entry + and move to destination. + + :param q_policy_index: the policy_index for the source object + :param account: the account name of the misplaced object + :param container: the container name of the misplaced object + :param obj: the name of the misplaced object + :param q_ts: a float, the timestamp of the misplaced object + :param path: the full path of the misplaced object for logging + :param container_policy_index: the policy_index of the destination + :param source_ts: a float, the timestamp of the source object + :param source_obj_status: the HTTP status source object request + :param source_obj_info: the HTTP headers of the source object request + :param source_obj_iter: the body iter of the source object request + """ + if source_obj_status // 100 != 2 or source_ts < q_ts: + if q_ts < time.time() - self.reclaim_age: + # it's old and there are no tombstones or anything; give up + self.stats_log('lost_source', '%r (%f) was not available in ' + 'policy_index %s and has expired', path, q_ts, + q_policy_index, level=logging.CRITICAL) + return True + # the source object is unavailable or older than the queue + # entry; a version that will satisfy the queue entry hopefully + # exists somewhere in the cluster, so wait and try again + self.stats_log('unavailable_source', '%r (%f) in ' + 'policy_index %s responded %s (%f)', path, + q_ts, q_policy_index, source_obj_status, + source_ts, level=logging.WARNING) + return False + + # optimistically move any source with a timestamp >= q_ts + ts = max(float(source_ts), q_ts) + # move the object + put_timestamp = slightly_later_timestamp(ts, offset=2) + self.stats_log('copy_attempt', '%r (%f) in policy_index %s will be ' + 'moved to policy_index %s (%s)', path, source_ts, + q_policy_index, container_policy_index, put_timestamp) + headers = source_obj_info.copy() + headers['X-Backend-Storage-Policy-Index'] = container_policy_index + headers['X-Timestamp'] = put_timestamp + + try: + self.swift.upload_object( + FileLikeIter(source_obj_iter), account, container, obj, + headers=headers) + except UnexpectedResponse as err: + self.stats_log('copy_failed', 'upload %r (%f) from ' + 'policy_index %s to policy_index %s ' + 'returned %s', path, source_ts, q_policy_index, + container_policy_index, err, level=logging.WARNING) + return False + except: # noqa + self.stats_log('unhandled_error', 'unable to upload %r (%f) ' + 'from policy_index %s to policy_index %s ', path, + source_ts, q_policy_index, container_policy_index, + level=logging.ERROR, exc_info=True) + return False + + self.stats_log('copy_success', '%r (%f) moved from policy_index %s ' + 'to policy_index %s (%s)', path, source_ts, + q_policy_index, container_policy_index, put_timestamp) + + return self.throw_tombstones(account, container, obj, q_ts, + q_policy_index, path) + + def ensure_tombstone_in_right_location(self, q_policy_index, account, + container, obj, q_ts, path, + container_policy_index, source_ts, + **kwargs): + """ + Issue a DELETE request against the destination to match the + misplaced DELETE against the source. + """ + delete_timestamp = slightly_later_timestamp(q_ts, offset=2) + self.stats_log('delete_attempt', '%r (%f) in policy_index %s ' + 'will be deleted from policy_index %s (%s)', path, + source_ts, q_policy_index, container_policy_index, + delete_timestamp) + headers = { + 'X-Backend-Storage-Policy-Index': container_policy_index, + 'X-Timestamp': delete_timestamp, + } + try: + self.swift.delete_object(account, container, obj, + headers=headers) + except UnexpectedResponse as err: + self.stats_log('delete_failed', 'delete %r (%f) from ' + 'policy_index %s (%s) returned %s', path, + source_ts, container_policy_index, + delete_timestamp, err, level=logging.WARNING) + return False + except: # noqa + self.stats_log('unhandled_error', 'unable to delete %r (%f) ' + 'from policy_index %s (%s)', path, source_ts, + container_policy_index, delete_timestamp, + level=logging.ERROR, exc_info=True) + return False + + self.stats_log('delete_success', '%r (%f) deleted from ' + 'policy_index %s (%s)', path, source_ts, + container_policy_index, delete_timestamp, + level=logging.INFO) + + return self.throw_tombstones(account, container, obj, q_ts, + q_policy_index, path) + + def reconcile_object(self, info): + """ + Process a possibly misplaced object write request. Determine correct + destination storage policy by checking with primary containers. Check + source and destination, copying or deleting into destination and + cleaning up the source as needed. + + This method wraps _reconcile_object for exception handling. + + :param info: a queue entry dict + + :returns: True to indicate the request is fully processed + successfully, otherwise False. + """ + self.logger.debug('checking placement for %r (%f) ' + 'in policy_index %s', info['path'], + info['q_ts'], info['q_policy_index']) + success = False + try: + success = self._reconcile_object(**info) + except: # noqa + self.logger.exception('Unhandled Exception trying to ' + 'reconcile %r (%s) in policy_index %s', + info['path'], info['q_ts'], + info['q_policy_index']) + if success: + metric = 'success' + msg = 'was handled successfully' + else: + metric = 'retry' + msg = 'must be retried' + msg = '%(path)r (%(q_ts)f) in policy_index %(q_policy_index)s ' + msg + self.stats_log(metric, msg, info, level=logging.INFO) + self.log_stats() + return success + + def _iter_containers(self): + """ + Generate a list of containers to process. + """ + # hit most recent container first instead of waiting on the updaters + current_container = get_reconciler_container_name(time.time()) + yield current_container + container_gen = self.swift.iter_containers(MISPLACED_OBJECTS_ACCOUNT) + self.logger.debug('looking for containers in %s', + MISPLACED_OBJECTS_ACCOUNT) + while True: + one_page = None + try: + one_page = list(itertools.islice( + container_gen, constraints.CONTAINER_LISTING_LIMIT)) + except UnexpectedResponse as err: + self.logger.error('Error listing containers in ' + 'account %s (%s)', + MISPLACED_OBJECTS_ACCOUNT, err) + + if not one_page: + # don't generally expect more than one page + break + # reversed order since we expect older containers to be empty + for c in reversed(one_page): + # encoding here is defensive + container = c['name'].encode('utf8') + if container == current_container: + continue # we've already hit this one this pass + yield container + + def _iter_objects(self, container): + """ + Generate a list of objects to process. + + :param container: the name of the container to process + + If the given container is empty and older than reclaim_age this + processor will attempt to reap it. + """ + self.logger.debug('looking for objects in %s', container) + found_obj = False + try: + for raw_obj in self.swift.iter_objects( + MISPLACED_OBJECTS_ACCOUNT, container): + found_obj = True + yield raw_obj + except UnexpectedResponse as err: + self.logger.error('Error listing objects in container %s (%s)', + container, err) + if float(container) < time.time() - self.reclaim_age and \ + not found_obj: + # Try to delete old empty containers so the queue doesn't + # grow without bound. It's ok if there's a conflict. + self.swift.delete_container( + MISPLACED_OBJECTS_ACCOUNT, container, + acceptable_statuses=(2, 404, 409, 412)) + + def reconcile(self): + """ + Main entry point for processing misplaced objects. + + Iterate over all queue entries and delegate to reconcile_object. + """ + self.logger.debug('pulling items from the queue') + for container in self._iter_containers(): + for raw_obj in self._iter_objects(container): + try: + obj_info = parse_raw_obj(raw_obj) + except Exception: + self.stats_log('invalid_record', + 'invalid queue record: %r', raw_obj, + level=logging.ERROR, exc_info=True) + continue + finished = self.reconcile_object(obj_info) + if finished: + self.pop_queue(container, raw_obj['name'], + obj_info['q_ts'], + obj_info['q_record']) + self.log_stats() + self.logger.debug('finished container %s', container) + + def run_once(self, *args, **kwargs): + """ + Process every entry in the queue. + """ + try: + self.reconcile() + except: + self.logger.exception('Unhandled Exception trying to reconcile') + self.log_stats(force=True) + + def run_forever(self, *args, **kwargs): + while True: + self.run_once(*args, **kwargs) + self.stats = defaultdict(int) + self.logger.info('sleeping between intervals (%ss)', self.interval) + time.sleep(self.interval) diff --git a/test/unit/__init__.py b/test/unit/__init__.py index d44ee7346c..ba155b8a41 100644 --- a/test/unit/__init__.py +++ b/test/unit/__init__.py @@ -626,7 +626,10 @@ def fake_http_connect(*code_iter, **kwargs): self._next_sleep = None def getresponse(self): - if kwargs.get('raise_exc'): + exc = kwargs.get('raise_exc') + if exc: + if isinstance(exc, Exception): + raise exc raise Exception('test') if kwargs.get('raise_timeout_exc'): raise Timeout() diff --git a/test/unit/common/middleware/helpers.py b/test/unit/common/middleware/helpers.py index a9f1d993c6..68a4bfee3d 100644 --- a/test/unit/common/middleware/helpers.py +++ b/test/unit/common/middleware/helpers.py @@ -20,6 +20,8 @@ from hashlib import md5 from swift.common import swob from swift.common.utils import split_path +from test.unit import FakeLogger, FakeRing + class FakeSwift(object): """ @@ -33,6 +35,21 @@ class FakeSwift(object): self.uploaded = {} # mapping of (method, path) --> (response class, headers, body) self._responses = {} + self.logger = FakeLogger('fake-swift') + self.account_ring = FakeRing() + self.container_ring = FakeRing() + self.get_object_ring = lambda policy_index: FakeRing() + + def _get_response(self, method, path): + resp = self._responses[(method, path)] + if isinstance(resp, list): + try: + resp = resp.pop(0) + except IndexError: + raise IndexError("Didn't find any more %r " + "in allowed responses" % ( + (method, path),)) + return resp def __call__(self, env, start_response): method = env['REQUEST_METHOD'] @@ -47,29 +64,30 @@ class FakeSwift(object): if resp: return resp(env, start_response) - headers = swob.Request(env).headers - self._calls.append((method, path, headers)) + req_headers = swob.Request(env).headers self.swift_sources.append(env.get('swift.source')) try: - resp_class, raw_headers, body = self._responses[(method, path)] + resp_class, raw_headers, body = self._get_response(method, path) headers = swob.HeaderKeyDict(raw_headers) except KeyError: if (env.get('QUERY_STRING') and (method, env['PATH_INFO']) in self._responses): - resp_class, raw_headers, body = self._responses[ - (method, env['PATH_INFO'])] + resp_class, raw_headers, body = self._get_response( + method, env['PATH_INFO']) headers = swob.HeaderKeyDict(raw_headers) elif method == 'HEAD' and ('GET', path) in self._responses: - resp_class, raw_headers, _ = self._responses[('GET', path)] + resp_class, raw_headers, body = self._get_response('GET', path) body = None headers = swob.HeaderKeyDict(raw_headers) elif method == 'GET' and obj and path in self.uploaded: resp_class = swob.HTTPOk headers, body = self.uploaded[path] else: - print "Didn't find %r in allowed responses" % ((method, path),) - raise + raise KeyError("Didn't find %r in allowed responses" % ( + (method, path),)) + + self._calls.append((method, path, req_headers)) # simulate object PUT if method == 'PUT' and obj: @@ -93,6 +111,10 @@ class FakeSwift(object): def calls(self): return [(method, path) for method, path, headers in self._calls] + @property + def headers(self): + return [headers for method, path, headers in self._calls] + @property def calls_with_headers(self): return self._calls @@ -103,3 +125,6 @@ class FakeSwift(object): def register(self, method, path, response_class, headers, body=''): self._responses[(method, path)] = (response_class, headers, body) + + def register_responses(self, method, path, responses): + self._responses[(method, path)] = list(responses) diff --git a/test/unit/common/test_utils.py b/test/unit/common/test_utils.py index 79a20c5fdd..712d5653e8 100644 --- a/test/unit/common/test_utils.py +++ b/test/unit/common/test_utils.py @@ -264,6 +264,16 @@ class TestUtils(unittest.TestCase): self.assertRaises(ValueError, utils.normalize_timestamp, '') self.assertRaises(ValueError, utils.normalize_timestamp, 'abc') + def test_last_modified_date_to_timestamp(self): + expectations = { + '1970-01-01T00:00:00.000000': 0.0, + '2014-02-28T23:22:36.698390': 1393629756.698390, + '2011-03-19T04:03:00.604554': 1300507380.604554, + } + for last_modified, ts in expectations.items(): + real = utils.last_modified_date_to_timestamp(last_modified) + self.assertEqual(real, ts, "failed for %s" % last_modified) + def test_backwards(self): # Test swift.common.utils.backward diff --git a/test/unit/container/test_reconciler.py b/test/unit/container/test_reconciler.py new file mode 100644 index 0000000000..f49cdfdb37 --- /dev/null +++ b/test/unit/container/test_reconciler.py @@ -0,0 +1,1697 @@ +# 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 json +import contextlib +import mock +import operator +import time +import unittest +import urllib +import socket +import os +import errno +import itertools +import random + +from collections import defaultdict +from datetime import datetime +from swift.container import reconciler +from swift.container.server import gen_resp_headers +from swift.common.direct_client import ClientException +from swift.common import swob +from swift.common.utils import split_path, normalize_timestamp + +from test.unit import debug_logger, FakeRing, fake_http_connect +from test.unit.common.middleware.helpers import FakeSwift + + +def timestamp_to_last_modified(timestamp): + return datetime.fromtimestamp(timestamp).strftime('%Y-%m-%dT%H:%M:%S.%f') + + +def container_resp_headers(**kwargs): + return swob.HeaderKeyDict(gen_resp_headers(kwargs)) + + +class FakeStoragePolicySwift(object): + + def __init__(self): + self.storage_policy = defaultdict(FakeSwift) + self._mock_oldest_spi_map = {} + + def __getattribute__(self, name): + try: + return object.__getattribute__(self, name) + except AttributeError: + return getattr(self.storage_policy[None], name) + + def __call__(self, env, start_response): + method = env['REQUEST_METHOD'] + path = env['PATH_INFO'] + _, acc, cont, obj = split_path(env['PATH_INFO'], 0, 4, + rest_with_last=True) + if not obj: + policy_index = None + else: + policy_index = self._mock_oldest_spi_map.get(cont, 0) + # allow backend policy override + if 'HTTP_X_BACKEND_STORAGE_POLICY_INDEX' in env: + policy_index = int(env['HTTP_X_BACKEND_STORAGE_POLICY_INDEX']) + + try: + return self.storage_policy[policy_index].__call__( + env, start_response) + except KeyError: + pass + + if method == 'PUT': + resp_class = swob.HTTPCreated + else: + resp_class = swob.HTTPNotFound + self.storage_policy[policy_index].register( + method, path, resp_class, {}, '') + + return self.storage_policy[policy_index].__call__( + env, start_response) + + +class FakeInternalClient(reconciler.InternalClient): + def __init__(self, listings): + self.app = FakeStoragePolicySwift() + self.user_agent = 'fake-internal-client' + self.request_tries = 1 + self.parse(listings) + + def parse(self, listings): + self.accounts = defaultdict(lambda: defaultdict(list)) + for item, timestamp in listings.items(): + # XXX this interface is stupid + if isinstance(timestamp, tuple): + timestamp, content_type = timestamp + else: + timestamp, content_type = timestamp, 'application/x-put' + storage_policy_index, path = item + account, container_name, obj_name = split_path( + path.encode('utf-8'), 0, 3, rest_with_last=True) + self.accounts[account][container_name].append( + (obj_name, storage_policy_index, timestamp, content_type)) + for account_name, containers in self.accounts.items(): + for con in containers: + self.accounts[account_name][con].sort(key=lambda t: t[0]) + for account, containers in self.accounts.items(): + account_listing_data = [] + account_path = '/v1/%s' % account + for container, objects in containers.items(): + container_path = account_path + '/' + container + container_listing_data = [] + for entry in objects: + (obj_name, storage_policy_index, + timestamp, content_type) = entry + if storage_policy_index is None and not obj_name: + # empty container + continue + obj_path = container_path + '/' + obj_name + headers = {'X-Timestamp': normalize_timestamp(timestamp)} + # register object response + self.app.storage_policy[storage_policy_index].register( + 'GET', obj_path, swob.HTTPOk, headers) + self.app.storage_policy[storage_policy_index].register( + 'DELETE', obj_path, swob.HTTPNoContent, {}) + # container listing entry + last_modified = timestamp_to_last_modified( + float(timestamp)) + obj_data = { + 'bytes': 0, + # listing data is unicode + 'name': obj_name.decode('utf-8'), + 'last_modified': last_modified, + 'hash': timestamp, + 'content_type': content_type, + } + container_listing_data.append(obj_data) + container_listing_data.sort(key=operator.itemgetter('name')) + # register container listing response + container_headers = {} + container_qry_string = '?format=json&marker=&end_marker=' + self.app.register('GET', container_path + container_qry_string, + swob.HTTPOk, container_headers, + json.dumps(container_listing_data)) + if container_listing_data: + obj_name = container_listing_data[-1]['name'] + # client should quote and encode marker + end_qry_string = '?format=json&marker=%s&end_marker=' % ( + urllib.quote(obj_name.encode('utf-8'))) + self.app.register('GET', container_path + end_qry_string, + swob.HTTPOk, container_headers, + json.dumps([])) + self.app.register('DELETE', container_path, + swob.HTTPConflict, {}, '') + # simple account listing entry + container_data = {'name': container} + account_listing_data.append(container_data) + # register account response + account_listing_data.sort(key=operator.itemgetter('name')) + account_headers = {} + account_qry_string = '?format=json&marker=&end_marker=' + self.app.register('GET', account_path + account_qry_string, + swob.HTTPOk, account_headers, + json.dumps(account_listing_data)) + end_qry_string = '?format=json&marker=%s&end_marker=' % ( + urllib.quote(account_listing_data[-1]['name'])) + self.app.register('GET', account_path + end_qry_string, + swob.HTTPOk, account_headers, + json.dumps([])) + + +class TestReconcilerUtils(unittest.TestCase): + + def setUp(self): + self.fake_ring = FakeRing() + + def test_parse_raw_obj(self): + got = reconciler.parse_raw_obj({ + 'name': "2:/AUTH_bob/con/obj", + 'hash': normalize_timestamp(2017551.49350), + 'last_modified': timestamp_to_last_modified(2017551.49352), + 'content_type': 'application/x-delete', + }) + self.assertEqual(got['q_policy_index'], 2) + self.assertEqual(got['account'], 'AUTH_bob') + self.assertEqual(got['container'], 'con') + self.assertEqual(got['obj'], 'obj') + self.assertEqual(got['q_ts'], 2017551.49350) + self.assertEqual(got['q_record'], 2017551.49352) + self.assertEqual(got['q_op'], 'DELETE') + + got = reconciler.parse_raw_obj({ + 'name': "1:/AUTH_bob/con/obj", + 'hash': normalize_timestamp(1234.20190), + 'last_modified': timestamp_to_last_modified(1234.20192), + 'content_type': 'application/x-put', + }) + self.assertEqual(got['q_policy_index'], 1) + self.assertEqual(got['account'], 'AUTH_bob') + self.assertEqual(got['container'], 'con') + self.assertEqual(got['obj'], 'obj') + self.assertEqual(got['q_ts'], 0000001234.20190) + self.assertEqual(got['q_record'], 0000001234.20192) + self.assertEqual(got['q_op'], 'PUT') + + # negative test + obj_info = { + 'name': "1:/AUTH_bob/con/obj", + 'hash': normalize_timestamp(1234.20190), + 'last_modified': timestamp_to_last_modified(1234.20192), + } + self.assertRaises(ValueError, reconciler.parse_raw_obj, obj_info) + obj_info['content_type'] = 'foo' + self.assertRaises(ValueError, reconciler.parse_raw_obj, obj_info) + obj_info['content_type'] = 'appliation/x-post' + self.assertRaises(ValueError, reconciler.parse_raw_obj, obj_info) + self.assertRaises(ValueError, reconciler.parse_raw_obj, + {'name': 'bogus'}) + self.assertRaises(ValueError, reconciler.parse_raw_obj, + {'name': '-1:/AUTH_test/container'}) + self.assertRaises(ValueError, reconciler.parse_raw_obj, + {'name': 'asdf:/AUTH_test/c/obj'}) + self.assertRaises(KeyError, reconciler.parse_raw_obj, + {'name': '0:/AUTH_test/c/obj', + 'content_type': 'application/x-put'}) + + def test_get_container_policy_index(self): + ts = itertools.count(int(time.time())) + mock_path = 'swift.container.reconciler.direct_head_container' + stub_resp_headers = [ + container_resp_headers( + status_changed_at=normalize_timestamp(ts.next()), + storage_policy_index=0, + ), + container_resp_headers( + status_changed_at=normalize_timestamp(ts.next()), + storage_policy_index=1, + ), + container_resp_headers( + status_changed_at=normalize_timestamp(ts.next()), + storage_policy_index=0, + ), + ] + for permutation in itertools.permutations((0, 1, 2)): + resp_headers = [stub_resp_headers[i] for i in permutation] + with mock.patch(mock_path) as direct_head: + direct_head.side_effect = resp_headers + oldest_spi = reconciler.direct_get_container_policy_index( + self.fake_ring, 'a', 'con') + test_values = [(info['x-storage-policy-index'], + info['x-backend-status-changed-at']) for + info in resp_headers] + self.assertEqual(oldest_spi, 0, + "oldest policy index wrong " + "for permutation %r" % test_values) + + def test_get_container_policy_index_with_error(self): + ts = itertools.count(int(time.time())) + mock_path = 'swift.container.reconciler.direct_head_container' + stub_resp_headers = [ + container_resp_headers( + status_change_at=ts.next(), + storage_policy_index=2, + ), + container_resp_headers( + status_changed_at=ts.next(), + storage_policy_index=1, + ), + # old timestamp, but 500 should be ignored... + ClientException( + 'Container Server blew up', + http_status=500, http_reason='Server Error', + http_headers=container_resp_headers( + status_changed_at=normalize_timestamp(0), + storage_policy_index=0, + ), + ), + ] + random.shuffle(stub_resp_headers) + with mock.patch(mock_path) as direct_head: + direct_head.side_effect = stub_resp_headers + oldest_spi = reconciler.direct_get_container_policy_index( + self.fake_ring, 'a', 'con') + self.assertEqual(oldest_spi, 2) + + def test_get_container_policy_index_with_socket_error(self): + ts = itertools.count(int(time.time())) + mock_path = 'swift.container.reconciler.direct_head_container' + stub_resp_headers = [ + container_resp_headers( + status_changed_at=normalize_timestamp(ts.next()), + storage_policy_index=1, + ), + container_resp_headers( + status_changed_at=normalize_timestamp(ts.next()), + storage_policy_index=0, + ), + socket.error(errno.ECONNREFUSED, os.strerror(errno.ECONNREFUSED)), + ] + random.shuffle(stub_resp_headers) + with mock.patch(mock_path) as direct_head: + direct_head.side_effect = stub_resp_headers + oldest_spi = reconciler.direct_get_container_policy_index( + self.fake_ring, 'a', 'con') + self.assertEqual(oldest_spi, 1) + + def test_get_container_policy_index_with_too_many_errors(self): + ts = itertools.count(int(time.time())) + mock_path = 'swift.container.reconciler.direct_head_container' + stub_resp_headers = [ + container_resp_headers( + status_changed_at=normalize_timestamp(ts.next()), + storage_policy_index=0, + ), + socket.error(errno.ECONNREFUSED, os.strerror(errno.ECONNREFUSED)), + ClientException( + 'Container Server blew up', + http_status=500, http_reason='Server Error', + http_headers=container_resp_headers( + status_changed_at=normalize_timestamp(ts.next()), + storage_policy_index=1, + ), + ), + ] + random.shuffle(stub_resp_headers) + with mock.patch(mock_path) as direct_head: + direct_head.side_effect = stub_resp_headers + oldest_spi = reconciler.direct_get_container_policy_index( + self.fake_ring, 'a', 'con') + self.assertEqual(oldest_spi, None) + + def test_get_container_policy_index_for_deleted(self): + mock_path = 'swift.container.reconciler.direct_head_container' + headers = container_resp_headers( + status_changed_at=normalize_timestamp(time.time()), + storage_policy_index=1, + ) + stub_resp_headers = [ + ClientException( + 'Container Not Found', + http_status=404, http_reason='Not Found', + http_headers=headers, + ), + ClientException( + 'Container Not Found', + http_status=404, http_reason='Not Found', + http_headers=headers, + ), + ClientException( + 'Container Not Found', + http_status=404, http_reason='Not Found', + http_headers=headers, + ), + ] + random.shuffle(stub_resp_headers) + with mock.patch(mock_path) as direct_head: + direct_head.side_effect = stub_resp_headers + oldest_spi = reconciler.direct_get_container_policy_index( + self.fake_ring, 'a', 'con') + self.assertEqual(oldest_spi, 1) + + def test_get_container_policy_index_for_recently_deleted(self): + ts = itertools.count(int(time.time())) + mock_path = 'swift.container.reconciler.direct_head_container' + stub_resp_headers = [ + ClientException( + 'Container Not Found', + http_status=404, http_reason='Not Found', + http_headers=container_resp_headers( + put_timestamp=ts.next(), + delete_timestamp=ts.next(), + status_changed_at=ts.next(), + storage_policy_index=0, + ), + ), + ClientException( + 'Container Not Found', + http_status=404, http_reason='Not Found', + http_headers=container_resp_headers( + put_timestamp=ts.next(), + delete_timestamp=ts.next(), + status_changed_at=ts.next(), + storage_policy_index=1, + ), + ), + ClientException( + 'Container Not Found', + http_status=404, http_reason='Not Found', + http_headers=container_resp_headers( + put_timestamp=ts.next(), + delete_timestamp=ts.next(), + status_changed_at=ts.next(), + storage_policy_index=2, + ), + ), + ] + random.shuffle(stub_resp_headers) + with mock.patch(mock_path) as direct_head: + direct_head.side_effect = stub_resp_headers + oldest_spi = reconciler.direct_get_container_policy_index( + self.fake_ring, 'a', 'con') + self.assertEqual(oldest_spi, 2) + + def test_get_container_policy_index_for_recently_recreated(self): + ts = itertools.count(int(time.time())) + mock_path = 'swift.container.reconciler.direct_head_container' + stub_resp_headers = [ + # old put, no recreate + container_resp_headers( + delete_timestamp=0, + put_timestamp=ts.next(), + status_changed_at=ts.next(), + storage_policy_index=0, + ), + # recently deleted + ClientException( + 'Container Not Found', + http_status=404, http_reason='Not Found', + http_headers=container_resp_headers( + put_timestamp=ts.next(), + delete_timestamp=ts.next(), + status_changed_at=ts.next(), + storage_policy_index=1, + ), + ), + # recently recreated + container_resp_headers( + delete_timestamp=ts.next(), + put_timestamp=ts.next(), + status_changed_at=ts.next(), + storage_policy_index=2, + ), + ] + random.shuffle(stub_resp_headers) + with mock.patch(mock_path) as direct_head: + direct_head.side_effect = stub_resp_headers + oldest_spi = reconciler.direct_get_container_policy_index( + self.fake_ring, 'a', 'con') + self.assertEqual(oldest_spi, 2) + + def test_get_container_policy_index_for_recently_split_brain(self): + ts = itertools.count(int(time.time())) + mock_path = 'swift.container.reconciler.direct_head_container' + stub_resp_headers = [ + # oldest put + container_resp_headers( + delete_timestamp=0, + put_timestamp=ts.next(), + status_changed_at=ts.next(), + storage_policy_index=0, + ), + # old recreate + container_resp_headers( + delete_timestamp=ts.next(), + put_timestamp=ts.next(), + status_changed_at=ts.next(), + storage_policy_index=1, + ), + # recently put + container_resp_headers( + delete_timestamp=0, + put_timestamp=ts.next(), + status_changed_at=ts.next(), + storage_policy_index=2, + ), + ] + random.shuffle(stub_resp_headers) + with mock.patch(mock_path) as direct_head: + direct_head.side_effect = stub_resp_headers + oldest_spi = reconciler.direct_get_container_policy_index( + self.fake_ring, 'a', 'con') + self.assertEqual(oldest_spi, 1) + + def test_direct_delete_container_entry(self): + mock_path = 'swift.common.direct_client.http_connect' + connect_args = [] + + def test_connect(ipaddr, port, device, partition, method, path, + headers=None, query_string=None): + connect_args.append({ + 'ipaddr': ipaddr, 'port': port, 'device': device, + 'partition': partition, 'method': method, 'path': path, + 'headers': headers, 'query_string': query_string}) + + x_timestamp = normalize_timestamp(time.time()) + headers = {'x-timestamp': x_timestamp} + fake_hc = fake_http_connect(200, 200, 200, give_connect=test_connect) + with mock.patch(mock_path, fake_hc): + reconciler.direct_delete_container_entry( + self.fake_ring, 'a', 'c', 'o', headers=headers) + + self.assertEqual(len(connect_args), 3) + for args in connect_args: + self.assertEqual(args['method'], 'DELETE') + self.assertEqual(args['path'], '/a/c/o') + self.assertEqual(args['headers'].get('x-timestamp'), + headers['x-timestamp']) + + def test_direct_delete_container_entry_with_errors(self): + # setup mock direct_delete + mock_path = \ + 'swift.container.reconciler.direct_delete_container_object' + stub_resp = [ + None, + socket.error(errno.ECONNREFUSED, os.strerror(errno.ECONNREFUSED)), + ClientException( + 'Container Server blew up', + '10.0.0.12', 6001, 'sdj', 404, 'Not Found' + ), + ] + mock_direct_delete = mock.MagicMock() + mock_direct_delete.side_effect = stub_resp + + with contextlib.nested( + mock.patch(mock_path, mock_direct_delete), + mock.patch('eventlet.greenpool.DEBUG', False), + ): + rv = reconciler.direct_delete_container_entry( + self.fake_ring, 'a', 'c', 'o') + self.assertEqual(rv, None) + self.assertEqual(len(mock_direct_delete.mock_calls), 3) + + def test_add_to_reconciler_queue(self): + mock_path = 'swift.common.direct_client.http_connect' + connect_args = [] + + def test_connect(ipaddr, port, device, partition, method, path, + headers=None, query_string=None): + connect_args.append({ + 'ipaddr': ipaddr, 'port': port, 'device': device, + 'partition': partition, 'method': method, 'path': path, + 'headers': headers, 'query_string': query_string}) + + fake_hc = fake_http_connect(200, 200, 200, give_connect=test_connect) + with mock.patch(mock_path, fake_hc): + ret = reconciler.add_to_reconciler_queue( + self.fake_ring, 'a', 'c', 'o', 17, 5948918.63946, 'DELETE') + + self.assertTrue(ret) + self.assertEqual(ret, str(int(5948918.63946 // 3600 * 3600))) + self.assertEqual(len(connect_args), 3) + connect_args.sort(key=lambda a: (a['ipaddr'], a['port'])) + + required_headers = ('x-content-type', 'x-etag') + + for args in connect_args: + self.assertEqual(args['headers']['X-Timestamp'], '5948918.63946') + self.assertEqual(args['path'], + '/.misplaced_objects/5947200/17:/a/c/o') + self.assertEqual(args['headers']['X-Content-Type'], + 'application/x-delete') + for header in required_headers: + self.assert_(header in args['headers'], + '%r was missing request headers %r' % ( + header, args['headers'])) + + def test_add_to_reconciler_queue_force(self): + mock_path = 'swift.common.direct_client.http_connect' + connect_args = [] + + def test_connect(ipaddr, port, device, partition, method, path, + headers=None, query_string=None): + connect_args.append({ + 'ipaddr': ipaddr, 'port': port, 'device': device, + 'partition': partition, 'method': method, 'path': path, + 'headers': headers, 'query_string': query_string}) + + fake_hc = fake_http_connect(200, 200, 200, give_connect=test_connect) + now = float(normalize_timestamp(time.time())) + with contextlib.nested( + mock.patch(mock_path, fake_hc), + mock.patch('swift.container.reconciler.time.time', + lambda: now), + ): + ret = reconciler.add_to_reconciler_queue( + self.fake_ring, 'a', 'c', 'o', 17, 5948918.63946, 'PUT', + force=True) + + self.assertTrue(ret) + self.assertEqual(ret, str(int(5948918.63946 // 3600 * 3600))) + self.assertEqual(len(connect_args), 3) + connect_args.sort(key=lambda a: (a['ipaddr'], a['port'])) + + required_headers = ('x-size', 'x-content-type') + + for args in connect_args: + self.assertEqual(args['headers']['X-Timestamp'], + normalize_timestamp(now)) + self.assertEqual(args['headers']['X-Etag'], '5948918.63946') + self.assertEqual(args['path'], + '/.misplaced_objects/5947200/17:/a/c/o') + for header in required_headers: + self.assert_(header in args['headers'], + '%r was missing request headers %r' % ( + header, args['headers'])) + + def test_add_to_reconciler_queue_fails(self): + mock_path = 'swift.common.direct_client.http_connect' + + fake_connects = [fake_http_connect(200), + fake_http_connect(200, raise_timeout_exc=True), + fake_http_connect(507)] + + def fake_hc(*a, **kw): + return fake_connects.pop()(*a, **kw) + + with mock.patch(mock_path, fake_hc): + ret = reconciler.add_to_reconciler_queue( + self.fake_ring, 'a', 'c', 'o', 17, 5948918.63946, 'PUT') + self.assertFalse(ret) + + def test_add_to_reconciler_queue_socket_error(self): + mock_path = 'swift.common.direct_client.http_connect' + + exc = socket.error(errno.ECONNREFUSED, + os.strerror(errno.ECONNREFUSED)) + fake_connects = [fake_http_connect(200), + fake_http_connect(200, raise_timeout_exc=True), + fake_http_connect(500, raise_exc=exc)] + + def fake_hc(*a, **kw): + return fake_connects.pop()(*a, **kw) + + with mock.patch(mock_path, fake_hc): + ret = reconciler.add_to_reconciler_queue( + self.fake_ring, 'a', 'c', 'o', 17, 5948918.63946, 'DELETE') + self.assertFalse(ret) + + +def listing_qs(marker): + return "?format=json&marker=%s&end_marker=" % \ + urllib.quote(marker.encode('utf-8')) + + +class TestReconciler(unittest.TestCase): + + maxDiff = None + + def setUp(self): + self.logger = debug_logger() + conf = {} + with mock.patch('swift.container.reconciler.InternalClient'): + self.reconciler = reconciler.ContainerReconciler(conf) + self.reconciler.logger = self.logger + self.start_interval = int(time.time() // 3600 * 3600) + self.current_container_path = '/v1/.misplaced_objects/%d' % ( + self.start_interval) + listing_qs('') + + def _mock_listing(self, objects): + self.reconciler.swift = FakeInternalClient(objects) + self.fake_swift = self.reconciler.swift.app + + def _mock_oldest_spi(self, container_oldest_spi_map): + self.fake_swift._mock_oldest_spi_map = container_oldest_spi_map + + def _run_once(self): + """ + Helper method to run the reconciler once with appropriate direct-client + mocks in place. + + Returns the list of direct-deleted container entries in the format + [(acc1, con1, obj1), ...] + """ + + def mock_oldest_spi(ring, account, container_name): + return self.fake_swift._mock_oldest_spi_map.get(container_name, 0) + + items = { + 'direct_get_container_policy_index': mock_oldest_spi, + 'direct_delete_container_entry': mock.DEFAULT, + } + + mock_time_iter = itertools.count(self.start_interval) + with mock.patch.multiple(reconciler, **items) as mocks: + self.mock_delete_container_entry = \ + mocks['direct_delete_container_entry'] + with mock.patch('time.time', mock_time_iter.next): + self.reconciler.run_once() + + return [c[1][1:4] for c in + mocks['direct_delete_container_entry'].mock_calls] + + def test_invalid_queue_name(self): + self._mock_listing({ + (None, "/.misplaced_objects/3600/bogus"): 3618.84187, + }) + deleted_container_entries = self._run_once() + # we try to find something useful + self.assertEqual( + self.fake_swift.calls, + [('GET', self.current_container_path), + ('GET', '/v1/.misplaced_objects' + listing_qs('')), + ('GET', '/v1/.misplaced_objects' + listing_qs('3600')), + ('GET', '/v1/.misplaced_objects/3600' + listing_qs('')), + ('GET', '/v1/.misplaced_objects/3600' + + listing_qs('bogus'))]) + # but only get the bogus record + self.assertEqual(self.reconciler.stats['invalid_record'], 1) + # and just leave it on the queue + self.assertEqual(self.reconciler.stats['pop_queue'], 0) + self.assertFalse(deleted_container_entries) + + def test_invalid_queue_name_marches_onward(self): + # there's something useful there on the queue + self._mock_listing({ + (None, "/.misplaced_objects/3600/00000bogus"): 3600.0000, + (None, "/.misplaced_objects/3600/1:/AUTH_bob/c/o1"): 3618.84187, + (1, "/AUTH_bob/c/o1"): 3618.84187, + }) + self._mock_oldest_spi({'c': 1}) # already in the right spot! + deleted_container_entries = self._run_once() + # we get all the queue entries we can + self.assertEqual( + self.fake_swift.calls, + [('GET', self.current_container_path), + ('GET', '/v1/.misplaced_objects' + listing_qs('')), + ('GET', '/v1/.misplaced_objects' + listing_qs('3600')), + ('GET', '/v1/.misplaced_objects/3600' + listing_qs('')), + ('GET', '/v1/.misplaced_objects/3600' + + listing_qs('1:/AUTH_bob/c/o1'))]) + # and one is garbage + self.assertEqual(self.reconciler.stats['invalid_record'], 1) + # but the other is workable + self.assertEqual(self.reconciler.stats['noop_object'], 1) + # so pop the queue for that one + self.assertEqual(self.reconciler.stats['pop_queue'], 1) + self.assertEqual(deleted_container_entries, + [('.misplaced_objects', '3600', '1:/AUTH_bob/c/o1')]) + self.assertEqual(self.reconciler.stats['success'], 1) + + def test_queue_name_with_policy_index_delimiter_in_name(self): + q_path = '.misplaced_objects/3600' + obj_path = "AUTH_bob/c:sneaky/o1:sneaky" + # there's something useful there on the queue + self._mock_listing({ + (None, "/%s/1:/%s" % (q_path, obj_path)): 3618.84187, + (1, '/%s' % obj_path): 3618.84187, + }) + self._mock_oldest_spi({'c': 0}) + deleted_container_entries = self._run_once() + # we find the misplaced object + self.assertEqual(self.reconciler.stats['misplaced_object'], 1) + self.assertEqual( + self.fake_swift.calls, + [('GET', self.current_container_path), + ('GET', '/v1/.misplaced_objects' + listing_qs('')), + ('GET', '/v1/.misplaced_objects' + listing_qs('3600')), + ('GET', '/v1/.misplaced_objects/3600' + listing_qs('')), + ('GET', '/v1/.misplaced_objects/3600' + + listing_qs('1:/%s' % obj_path))]) + # move it + self.assertEqual(self.reconciler.stats['copy_attempt'], 1) + self.assertEqual(self.reconciler.stats['copy_success'], 1) + self.assertEqual( + self.fake_swift.storage_policy[1].calls, + [('GET', '/v1/%s' % obj_path), + ('DELETE', '/v1/%s' % obj_path)]) + delete_headers = self.fake_swift.storage_policy[1].headers[1] + self.assertEqual( + self.fake_swift.storage_policy[0].calls, + [('HEAD', '/v1/%s' % obj_path), + ('PUT', '/v1/%s' % obj_path)]) + # clean up the source + self.assertEqual(self.reconciler.stats['cleanup_attempt'], 1) + self.assertEqual(self.reconciler.stats['cleanup_success'], 1) + # we DELETE the object from the wrong place with source_ts + offset 1 + # timestamp to make sure the change takes effect + self.assertEqual(delete_headers.get('X-Timestamp'), + normalize_timestamp(3618.84188)) + # and pop the queue for that one + self.assertEqual(self.reconciler.stats['pop_queue'], 1) + self.assertEqual(deleted_container_entries, [( + '.misplaced_objects', '3600', '1:/%s' % obj_path)]) + self.assertEqual(self.reconciler.stats['success'], 1) + + def test_unable_to_direct_get_oldest_storage_policy(self): + self._mock_listing({ + (None, "/.misplaced_objects/3600/1:/AUTH_bob/c/o1"): 3618.84187, + }) + # the reconciler gets "None" if we can't quorum the container + self._mock_oldest_spi({'c': None}) + deleted_container_entries = self._run_once() + # we look for misplaced objects + self.assertEqual( + self.fake_swift.calls, + [('GET', self.current_container_path), + ('GET', '/v1/.misplaced_objects' + listing_qs('')), + ('GET', '/v1/.misplaced_objects' + listing_qs('3600')), + ('GET', '/v1/.misplaced_objects/3600' + listing_qs('')), + ('GET', '/v1/.misplaced_objects/3600' + + listing_qs('1:/AUTH_bob/c/o1'))]) + # but can't really say where to go looking + self.assertEqual(self.reconciler.stats['unavailable_container'], 1) + # we don't clean up anything + self.assertEqual(self.reconciler.stats['cleanup_object'], 0) + # and we definately should not pop_queue + self.assertFalse(deleted_container_entries) + self.assertEqual(self.reconciler.stats['retry'], 1) + + def test_object_move(self): + self._mock_listing({ + (None, "/.misplaced_objects/3600/1:/AUTH_bob/c/o1"): 3618.84187, + (1, "/AUTH_bob/c/o1"): 3618.84187, + }) + self._mock_oldest_spi({'c': 0}) + deleted_container_entries = self._run_once() + + # found a misplaced object + self.assertEqual(self.reconciler.stats['misplaced_object'], 1) + self.assertEqual( + self.fake_swift.calls, + [('GET', self.current_container_path), + ('GET', '/v1/.misplaced_objects' + listing_qs('')), + ('GET', '/v1/.misplaced_objects' + listing_qs('3600')), + ('GET', '/v1/.misplaced_objects/3600' + listing_qs('')), + ('GET', '/v1/.misplaced_objects/3600' + + listing_qs('1:/AUTH_bob/c/o1'))]) + # moves it + self.assertEqual(self.reconciler.stats['copy_attempt'], 1) + self.assertEqual(self.reconciler.stats['copy_success'], 1) + self.assertEqual( + self.fake_swift.storage_policy[1].calls, + [('GET', '/v1/AUTH_bob/c/o1'), + ('DELETE', '/v1/AUTH_bob/c/o1')]) + delete_headers = self.fake_swift.storage_policy[1].headers[1] + self.assertEqual( + self.fake_swift.storage_policy[0].calls, + [('HEAD', '/v1/AUTH_bob/c/o1'), + ('PUT', '/v1/AUTH_bob/c/o1')]) + put_headers = self.fake_swift.storage_policy[0].headers[1] + # we PUT the object in the right place with q_ts + offset 2 + self.assertEqual(put_headers.get('X-Timestamp'), + normalize_timestamp('3618.84189')) + # cleans up the old + self.assertEqual(self.reconciler.stats['cleanup_attempt'], 1) + self.assertEqual(self.reconciler.stats['cleanup_success'], 1) + # we DELETE the object from the wrong place with source_ts + offset 1 + # timestamp to make sure the change takes effect + self.assertEqual(delete_headers.get('X-Timestamp'), + normalize_timestamp('3618.84188')) + # and when we're done, we pop the entry from the queue + self.assertEqual(self.reconciler.stats['pop_queue'], 1) + self.assertEqual(deleted_container_entries, + [('.misplaced_objects', '3600', '1:/AUTH_bob/c/o1')]) + self.assertEqual(self.reconciler.stats['success'], 1) + + def test_object_move_the_other_direction(self): + self._mock_listing({ + (None, "/.misplaced_objects/3600/0:/AUTH_bob/c/o1"): 3618.84187, + (0, "/AUTH_bob/c/o1"): 3618.84187, + }) + self._mock_oldest_spi({'c': 1}) + deleted_container_entries = self._run_once() + + # found a misplaced object + self.assertEqual(self.reconciler.stats['misplaced_object'], 1) + self.assertEqual( + self.fake_swift.calls, + [('GET', self.current_container_path), + ('GET', '/v1/.misplaced_objects' + listing_qs('')), + ('GET', '/v1/.misplaced_objects' + listing_qs('3600')), + ('GET', '/v1/.misplaced_objects/3600' + listing_qs('')), + ('GET', '/v1/.misplaced_objects/3600' + + listing_qs('0:/AUTH_bob/c/o1'))]) + # moves it + self.assertEqual(self.reconciler.stats['copy_attempt'], 1) + self.assertEqual(self.reconciler.stats['copy_success'], 1) + self.assertEqual( + self.fake_swift.storage_policy[0].calls, + [('GET', '/v1/AUTH_bob/c/o1'), # 2 + ('DELETE', '/v1/AUTH_bob/c/o1')]) # 4 + delete_headers = self.fake_swift.storage_policy[0].headers[1] + self.assertEqual( + self.fake_swift.storage_policy[1].calls, + [('HEAD', '/v1/AUTH_bob/c/o1'), # 1 + ('PUT', '/v1/AUTH_bob/c/o1')]) # 3 + put_headers = self.fake_swift.storage_policy[1].headers[1] + # we PUT the object in the right place with q_ts + offset 2 + self.assertEqual(put_headers.get('X-Timestamp'), + normalize_timestamp('3618.84189')) + # cleans up the old + self.assertEqual(self.reconciler.stats['cleanup_attempt'], 1) + self.assertEqual(self.reconciler.stats['cleanup_success'], 1) + # we DELETE the object from the wrong place with source_ts + offset 1 + # timestamp to make sure the change takes effect + self.assertEqual(delete_headers.get('X-Timestamp'), + normalize_timestamp('3618.84188')) + # and when we're done, we pop the entry from the queue + self.assertEqual(self.reconciler.stats['pop_queue'], 1) + self.assertEqual(deleted_container_entries, + [('.misplaced_objects', '3600', '0:/AUTH_bob/c/o1')]) + self.assertEqual(self.reconciler.stats['success'], 1) + + def test_object_move_with_unicode_and_spaces(self): + # the "name" in listings and the unicode string passed to all + # functions where we call them with (account, container, obj) + obj_name = u"AUTH_bob/c \u062a/o1 \u062a" + # anytime we talk about a call made to swift for a path + obj_path = obj_name.encode('utf-8') + # this mock expects unquoted unicode because it handles container + # listings as well as paths + self._mock_listing({ + (None, "/.misplaced_objects/3600/1:/%s" % obj_name): 3618.84187, + (1, "/%s" % obj_name): 3618.84187, + }) + self._mock_oldest_spi({'c': 0}) + deleted_container_entries = self._run_once() + + # found a misplaced object + self.assertEqual(self.reconciler.stats['misplaced_object'], 1) + # listing_qs encodes and quotes - so give it name + self.assertEqual( + self.fake_swift.calls, + [('GET', self.current_container_path), + ('GET', '/v1/.misplaced_objects' + listing_qs('')), + ('GET', '/v1/.misplaced_objects' + listing_qs('3600')), + ('GET', '/v1/.misplaced_objects/3600' + listing_qs('')), + ('GET', '/v1/.misplaced_objects/3600' + + listing_qs('1:/%s' % obj_name))]) + # moves it + self.assertEqual(self.reconciler.stats['copy_attempt'], 1) + self.assertEqual(self.reconciler.stats['copy_success'], 1) + # these calls are to the real path + self.assertEqual( + self.fake_swift.storage_policy[1].calls, + [('GET', '/v1/%s' % obj_path), # 2 + ('DELETE', '/v1/%s' % obj_path)]) # 4 + delete_headers = self.fake_swift.storage_policy[1].headers[1] + self.assertEqual( + self.fake_swift.storage_policy[0].calls, + [('HEAD', '/v1/%s' % obj_path), # 1 + ('PUT', '/v1/%s' % obj_path)]) # 3 + put_headers = self.fake_swift.storage_policy[0].headers[1] + # we PUT the object in the right place with q_ts + offset 2 + self.assertEqual(put_headers.get('X-Timestamp'), + normalize_timestamp('3618.84189')) + # cleans up the old + self.assertEqual(self.reconciler.stats['cleanup_attempt'], 1) + self.assertEqual(self.reconciler.stats['cleanup_success'], 1) + # we DELETE the object from the wrong place with source_ts + offset 1 + # timestamp to make sure the change takes effect + self.assertEqual(delete_headers.get('X-Timestamp'), + normalize_timestamp('3618.84188')) + self.assertEqual( + delete_headers.get('X-Backend-Storage-Policy-Index'), '1') + # and when we're done, we pop the entry from the queue + self.assertEqual(self.reconciler.stats['pop_queue'], 1) + # this mock recieved the name, it's encoded down in buffered_http + self.assertEqual(deleted_container_entries, + [('.misplaced_objects', '3600', '1:/%s' % obj_name)]) + self.assertEqual(self.reconciler.stats['success'], 1) + + def test_object_delete(self): + q_ts = float(normalize_timestamp(time.time())) + self._mock_listing({ + (None, "/.misplaced_objects/3600/1:/AUTH_bob/c/o1"): ( + normalize_timestamp(q_ts), 'application/x-delete'), + # object exists in "correct" storage policy - slightly older + (0, "/AUTH_bob/c/o1"): normalize_timestamp(q_ts - 1), + }) + self._mock_oldest_spi({'c': 0}) + # the tombstone exists in the enqueued storage policy + self.fake_swift.storage_policy[1].register( + 'GET', '/v1/AUTH_bob/c/o1', swob.HTTPNotFound, + {'X-Backend-Timestamp': normalize_timestamp(q_ts)}) + deleted_container_entries = self._run_once() + + # found a misplaced object + self.assertEqual(self.reconciler.stats['misplaced_object'], 1) + self.assertEqual( + self.fake_swift.calls, + [('GET', self.current_container_path), + ('GET', '/v1/.misplaced_objects' + listing_qs('')), + ('GET', '/v1/.misplaced_objects' + listing_qs('3600')), + ('GET', '/v1/.misplaced_objects/3600' + listing_qs('')), + ('GET', '/v1/.misplaced_objects/3600' + + listing_qs('1:/AUTH_bob/c/o1'))]) + # delete it + self.assertEqual(self.reconciler.stats['delete_attempt'], 1) + self.assertEqual(self.reconciler.stats['delete_success'], 1) + self.assertEqual( + self.fake_swift.storage_policy[1].calls, + [('GET', '/v1/AUTH_bob/c/o1'), + ('DELETE', '/v1/AUTH_bob/c/o1')]) + delete_headers = self.fake_swift.storage_policy[1].headers[1] + self.assertEqual( + self.fake_swift.storage_policy[0].calls, + [('HEAD', '/v1/AUTH_bob/c/o1'), + ('DELETE', '/v1/AUTH_bob/c/o1')]) + reconcile_headers = self.fake_swift.storage_policy[0].headers[1] + # we DELETE the object in the right place with q_ts + offset 2 + self.assertEqual(reconcile_headers.get('X-Timestamp'), + normalize_timestamp(q_ts + 0.00002)) + # cleans up the old + self.assertEqual(self.reconciler.stats['cleanup_attempt'], 1) + self.assertEqual(self.reconciler.stats['cleanup_success'], 1) + # we DELETE the object from the wrong place with source_ts + offset 1 + # timestamp to make sure the change takes effect + self.assertEqual(delete_headers.get('X-Timestamp'), + normalize_timestamp(q_ts + 0.00001)) + # and when we're done, we pop the entry from the queue + self.assertEqual(self.reconciler.stats['pop_queue'], 1) + self.assertEqual(deleted_container_entries, + [('.misplaced_objects', '3600', '1:/AUTH_bob/c/o1')]) + self.assertEqual(self.reconciler.stats['success'], 1) + + def test_object_enqueued_for_the_correct_dest_noop(self): + self._mock_listing({ + (None, "/.misplaced_objects/3600/1:/AUTH_bob/c/o1"): 3618.84187, + (1, "/AUTH_bob/c/o1"): 3618.84187, + }) + self._mock_oldest_spi({'c': 1}) # already in the right spot! + deleted_container_entries = self._run_once() + + # nothing to see here + self.assertEqual(self.reconciler.stats['noop_object'], 1) + self.assertEqual( + self.fake_swift.calls, + [('GET', self.current_container_path), + ('GET', '/v1/.misplaced_objects' + listing_qs('')), + ('GET', '/v1/.misplaced_objects' + listing_qs('3600')), + ('GET', '/v1/.misplaced_objects/3600' + listing_qs('')), + ('GET', '/v1/.misplaced_objects/3600' + + listing_qs('1:/AUTH_bob/c/o1'))]) + # so we just pop the queue + self.assertEqual(self.reconciler.stats['pop_queue'], 1) + self.assertEqual(deleted_container_entries, + [('.misplaced_objects', '3600', '1:/AUTH_bob/c/o1')]) + self.assertEqual(self.reconciler.stats['success'], 1) + + def test_object_move_src_object_newer_than_queue_entry(self): + # setup the cluster + self._mock_listing({ + (None, "/.misplaced_objects/3600/1:/AUTH_bob/c/o1"): 3600.123456, + (1, '/AUTH_bob/c/o1'): 3600.234567, # slightly newer + }) + self._mock_oldest_spi({'c': 0}) # destination + # turn the crank + deleted_container_entries = self._run_once() + + # found a misplaced object + self.assertEqual(self.reconciler.stats['misplaced_object'], 1) + self.assertEqual( + self.fake_swift.calls, + [('GET', self.current_container_path), + ('GET', '/v1/.misplaced_objects' + listing_qs('')), + ('GET', '/v1/.misplaced_objects' + listing_qs('3600')), + ('GET', '/v1/.misplaced_objects/3600' + listing_qs('')), + ('GET', '/v1/.misplaced_objects/3600' + + listing_qs('1:/AUTH_bob/c/o1'))]) + # proceed with the move + self.assertEqual(self.reconciler.stats['copy_attempt'], 1) + self.assertEqual(self.reconciler.stats['copy_success'], 1) + self.assertEqual( + self.fake_swift.storage_policy[1].calls, + [('GET', '/v1/AUTH_bob/c/o1'), # 2 + ('DELETE', '/v1/AUTH_bob/c/o1')]) # 4 + delete_headers = self.fake_swift.storage_policy[1].headers[1] + self.assertEqual( + self.fake_swift.storage_policy[0].calls, + [('HEAD', '/v1/AUTH_bob/c/o1'), # 1 + ('PUT', '/v1/AUTH_bob/c/o1')]) # 3 + # .. with source timestamp + offset 2 + put_headers = self.fake_swift.storage_policy[0].headers[1] + self.assertEqual(put_headers.get('X-Timestamp'), + normalize_timestamp(3600.234587)) + # src object is cleaned up + self.assertEqual(self.reconciler.stats['cleanup_attempt'], 1) + self.assertEqual(self.reconciler.stats['cleanup_success'], 1) + # ... with q_ts + offset 1 + self.assertEqual(delete_headers.get('X-Timestamp'), + normalize_timestamp(3600.123466)) + # and queue is popped + self.assertEqual(self.reconciler.stats['pop_queue'], 1) + self.assertEqual(deleted_container_entries, + [('.misplaced_objects', '3600', '1:/AUTH_bob/c/o1')]) + self.assertEqual(self.reconciler.stats['success'], 1) + + def test_object_move_src_object_older_than_queue_entry(self): + # should be some sort of retry case + q_ts = float(normalize_timestamp(time.time())) + container = str(int(q_ts // 3600 * 3600)) + q_path = '.misplaced_objects/%s' % container + self._mock_listing({ + (None, "/%s/1:/AUTH_bob/c/o1" % q_path): q_ts, + (1, '/AUTH_bob/c/o1'): q_ts - 0.00001, # slightly older + }) + self._mock_oldest_spi({'c': 0}) + deleted_container_entries = self._run_once() + + # found a misplaced object + self.assertEqual(self.reconciler.stats['misplaced_object'], 1) + self.assertEqual( + self.fake_swift.calls, + [('GET', '/v1/%s' % q_path + listing_qs('')), + ('GET', '/v1/%s' % q_path + + listing_qs('1:/AUTH_bob/c/o1')), + ('GET', '/v1/.misplaced_objects' + listing_qs('')), + ('GET', '/v1/.misplaced_objects' + listing_qs(container))]) + self.assertEqual( + self.fake_swift.storage_policy[0].calls, + [('HEAD', '/v1/AUTH_bob/c/o1')]) + # but no object copy is attempted + self.assertEqual(self.reconciler.stats['unavailable_source'], 1) + self.assertEqual(self.reconciler.stats['copy_attempt'], 0) + self.assertEqual( + self.fake_swift.storage_policy[1].calls, + [('GET', '/v1/AUTH_bob/c/o1')]) + # src object is un-modified + self.assertEqual(self.reconciler.stats['cleanup_attempt'], 0) + # queue is un-changed, we'll have to retry + self.assertEqual(self.reconciler.stats['pop_queue'], 0) + self.assertEqual(deleted_container_entries, []) + self.assertEqual(self.reconciler.stats['retry'], 1) + + def test_src_object_unavailable_with_slightly_newer_tombstone(self): + # should be some sort of retry case + q_ts = float(normalize_timestamp(time.time())) + container = str(int(q_ts // 3600 * 3600)) + q_path = '.misplaced_objects/%s' % container + self._mock_listing({ + (None, "/%s/1:/AUTH_bob/c/o1" % q_path): q_ts, + }) + self._mock_oldest_spi({'c': 0}) + self.fake_swift.storage_policy[1].register( + 'GET', '/v1/AUTH_bob/c/o1', swob.HTTPNotFound, + {'X-Backend-Timestamp': normalize_timestamp(q_ts + 0.00002)}) + deleted_container_entries = self._run_once() + + # found a misplaced object + self.assertEqual(self.reconciler.stats['misplaced_object'], 1) + self.assertEqual( + self.fake_swift.calls, + [('GET', '/v1/%s' % q_path + listing_qs('')), + ('GET', '/v1/%s' % q_path + + listing_qs('1:/AUTH_bob/c/o1')), + ('GET', '/v1/.misplaced_objects' + listing_qs('')), + ('GET', '/v1/.misplaced_objects' + listing_qs(container))]) + self.assertEqual( + self.fake_swift.storage_policy[0].calls, + [('HEAD', '/v1/AUTH_bob/c/o1')]) + # but no object copy is attempted + self.assertEqual(self.reconciler.stats['unavailable_source'], 1) + self.assertEqual(self.reconciler.stats['copy_attempt'], 0) + self.assertEqual( + self.fake_swift.storage_policy[1].calls, + [('GET', '/v1/AUTH_bob/c/o1')]) + # src object is un-modified + self.assertEqual(self.reconciler.stats['cleanup_attempt'], 0) + # queue is un-changed, we'll have to retry + self.assertEqual(self.reconciler.stats['pop_queue'], 0) + self.assertEqual(deleted_container_entries, []) + self.assertEqual(self.reconciler.stats['retry'], 1) + + def test_src_object_unavailable_server_error(self): + # should be some sort of retry case + q_ts = float(normalize_timestamp(time.time())) + container = str(int(q_ts // 3600 * 3600)) + q_path = '.misplaced_objects/%s' % container + self._mock_listing({ + (None, "/%s/1:/AUTH_bob/c/o1" % q_path): q_ts, + }) + self._mock_oldest_spi({'c': 0}) + self.fake_swift.storage_policy[1].register( + 'GET', '/v1/AUTH_bob/c/o1', swob.HTTPServiceUnavailable, {}) + deleted_container_entries = self._run_once() + + # found a misplaced object + self.assertEqual(self.reconciler.stats['misplaced_object'], 1) + self.assertEqual( + self.fake_swift.calls, + [('GET', '/v1/%s' % q_path + listing_qs('')), + ('GET', '/v1/%s' % q_path + + listing_qs('1:/AUTH_bob/c/o1')), + ('GET', '/v1/.misplaced_objects' + listing_qs('')), + ('GET', '/v1/.misplaced_objects' + listing_qs(container))]) + self.assertEqual( + self.fake_swift.storage_policy[0].calls, + [('HEAD', '/v1/AUTH_bob/c/o1')]) + # but no object copy is attempted + self.assertEqual(self.reconciler.stats['unavailable_source'], 1) + self.assertEqual(self.reconciler.stats['copy_attempt'], 0) + self.assertEqual( + self.fake_swift.storage_policy[1].calls, + [('GET', '/v1/AUTH_bob/c/o1')]) + # src object is un-modified + self.assertEqual(self.reconciler.stats['cleanup_attempt'], 0) + # queue is un-changed, we'll have to retry + self.assertEqual(self.reconciler.stats['pop_queue'], 0) + self.assertEqual(deleted_container_entries, []) + self.assertEqual(self.reconciler.stats['retry'], 1) + + def test_object_move_fails_cleanup(self): + # setup the cluster + self._mock_listing({ + (None, "/.misplaced_objects/3600/1:/AUTH_bob/c/o1"): 3600.123456, + (1, '/AUTH_bob/c/o1'): 3600.234567, # slightly newer + }) + self._mock_oldest_spi({'c': 0}) # destination + + # make the DELETE blow up + self.fake_swift.storage_policy[1].register( + 'DELETE', '/v1/AUTH_bob/c/o1', swob.HTTPServiceUnavailable, {}) + # turn the crank + deleted_container_entries = self._run_once() + + # found a misplaced object + self.assertEqual(self.reconciler.stats['misplaced_object'], 1) + self.assertEqual( + self.fake_swift.calls, + [('GET', self.current_container_path), + ('GET', '/v1/.misplaced_objects' + listing_qs('')), + ('GET', '/v1/.misplaced_objects' + listing_qs('3600')), + ('GET', '/v1/.misplaced_objects/3600' + listing_qs('')), + ('GET', '/v1/.misplaced_objects/3600' + + listing_qs('1:/AUTH_bob/c/o1'))]) + # proceed with the move + self.assertEqual(self.reconciler.stats['copy_attempt'], 1) + self.assertEqual(self.reconciler.stats['copy_success'], 1) + self.assertEqual( + self.fake_swift.storage_policy[1].calls, + [('GET', '/v1/AUTH_bob/c/o1'), # 2 + ('DELETE', '/v1/AUTH_bob/c/o1')]) # 4 + delete_headers = self.fake_swift.storage_policy[1].headers[1] + self.assertEqual( + self.fake_swift.storage_policy[0].calls, + [('HEAD', '/v1/AUTH_bob/c/o1'), # 1 + ('PUT', '/v1/AUTH_bob/c/o1')]) # 3 + # .. with source timestamp + offset 2 + put_headers = self.fake_swift.storage_policy[0].headers[1] + self.assertEqual(put_headers.get('X-Timestamp'), + normalize_timestamp(3600.234587)) + # we try to cleanup + self.assertEqual(self.reconciler.stats['cleanup_attempt'], 1) + # ... with q_ts + offset 1 + self.assertEqual(delete_headers.get('X-Timestamp'), + normalize_timestamp(3600.123466)) + # but cleanup fails! + self.assertEqual(self.reconciler.stats['cleanup_failed'], 1) + # so the queue is not popped + self.assertEqual(self.reconciler.stats['pop_queue'], 0) + self.assertEqual(deleted_container_entries, []) + # and we'll have to retry + self.assertEqual(self.reconciler.stats['retry'], 1) + + def test_object_move_src_object_is_forever_gone(self): + # oh boy, hate to be here - this is an oldy + q_ts = self.start_interval - self.reconciler.reclaim_age - 1 + self._mock_listing({ + (None, "/.misplaced_objects/3600/1:/AUTH_bob/c/o1"): q_ts, + }) + self._mock_oldest_spi({'c': 0}) + deleted_container_entries = self._run_once() + + # found a misplaced object + self.assertEqual(self.reconciler.stats['misplaced_object'], 1) + self.assertEqual( + self.fake_swift.calls, + [('GET', self.current_container_path), + ('GET', '/v1/.misplaced_objects' + listing_qs('')), + ('GET', '/v1/.misplaced_objects' + listing_qs('3600')), + ('GET', '/v1/.misplaced_objects/3600' + listing_qs('')), + ('GET', '/v1/.misplaced_objects/3600' + + listing_qs('1:/AUTH_bob/c/o1'))]) + self.assertEqual( + self.fake_swift.storage_policy[0].calls, + [('HEAD', '/v1/AUTH_bob/c/o1')]) + # but it's gone :\ + self.assertEqual(self.reconciler.stats['lost_source'], 1) + self.assertEqual( + self.fake_swift.storage_policy[1].calls, + [('GET', '/v1/AUTH_bob/c/o1')]) + # gah, look, even if it was out there somewhere - we've been at this + # two weeks and haven't found it. We can't just keep looking forever, + # so... we're done + self.assertEqual(self.reconciler.stats['pop_queue'], 1) + self.assertEqual(deleted_container_entries, + [('.misplaced_objects', '3600', '1:/AUTH_bob/c/o1')]) + # dunno if this is helpful, but FWIW we don't throw tombstones? + self.assertEqual(self.reconciler.stats['cleanup_attempt'], 0) + self.assertEqual(self.reconciler.stats['success'], 1) # lol + + def test_object_move_dest_already_moved(self): + self._mock_listing({ + (None, "/.misplaced_objects/3600/1:/AUTH_bob/c/o1"): 3679.2019, + (1, "/AUTH_bob/c/o1"): 3679.2019, + (0, "/AUTH_bob/c/o1"): 3679.2019, + }) + self._mock_oldest_spi({'c': 0}) + deleted_container_entries = self._run_once() + + # we look for misplaced objects + self.assertEqual( + self.fake_swift.calls, + [('GET', self.current_container_path), + ('GET', '/v1/.misplaced_objects' + listing_qs('')), + ('GET', '/v1/.misplaced_objects' + listing_qs('3600')), + ('GET', '/v1/.misplaced_objects/3600' + listing_qs('')), + ('GET', '/v1/.misplaced_objects/3600' + + listing_qs('1:/AUTH_bob/c/o1'))]) + # but we found it already in the right place! + self.assertEqual(self.reconciler.stats['found_object'], 1) + self.assertEqual( + self.fake_swift.storage_policy[0].calls, + [('HEAD', '/v1/AUTH_bob/c/o1')]) + # so no attempt to read the source is made, but we do cleanup + self.assertEqual( + self.fake_swift.storage_policy[1].calls, + [('DELETE', '/v1/AUTH_bob/c/o1')]) + delete_headers = self.fake_swift.storage_policy[1].headers[0] + # rather we just clean up the dark matter + self.assertEqual(self.reconciler.stats['cleanup_attempt'], 1) + self.assertEqual(self.reconciler.stats['cleanup_success'], 1) + self.assertEqual(delete_headers.get('X-Timestamp'), + normalize_timestamp(3679.20191)) + # and wipe our hands of it + self.assertEqual(self.reconciler.stats['pop_queue'], 1) + self.assertEqual(deleted_container_entries, + [('.misplaced_objects', '3600', '1:/AUTH_bob/c/o1')]) + self.assertEqual(self.reconciler.stats['success'], 1) + + def test_object_move_dest_object_newer_than_queue_entry(self): + self._mock_listing({ + (None, "/.misplaced_objects/3600/1:/AUTH_bob/c/o1"): 3679.2019, + (1, "/AUTH_bob/c/o1"): 3679.2019, + (0, "/AUTH_bob/c/o1"): 3679.2019 + 0.00001, # slightly newer + }) + self._mock_oldest_spi({'c': 0}) + deleted_container_entries = self._run_once() + + # we look for misplaced objects... + self.assertEqual( + self.fake_swift.calls, + [('GET', self.current_container_path), + ('GET', '/v1/.misplaced_objects' + listing_qs('')), + ('GET', '/v1/.misplaced_objects' + listing_qs('3600')), + ('GET', '/v1/.misplaced_objects/3600' + listing_qs('')), + ('GET', '/v1/.misplaced_objects/3600' + + listing_qs('1:/AUTH_bob/c/o1'))]) + # but we found it already in the right place! + self.assertEqual(self.reconciler.stats['found_object'], 1) + self.assertEqual( + self.fake_swift.storage_policy[0].calls, + [('HEAD', '/v1/AUTH_bob/c/o1')]) + # so not attempt to read is made, but we do cleanup + self.assertEqual(self.reconciler.stats['copy_attempt'], 0) + self.assertEqual( + self.fake_swift.storage_policy[1].calls, + [('DELETE', '/v1/AUTH_bob/c/o1')]) + delete_headers = self.fake_swift.storage_policy[1].headers[0] + # rather we just clean up the dark matter + self.assertEqual(self.reconciler.stats['cleanup_attempt'], 1) + self.assertEqual(self.reconciler.stats['cleanup_success'], 1) + + self.assertEqual(delete_headers.get('X-Timestamp'), + normalize_timestamp('3679.20191')) + # and since we cleaned up the old object, so this counts as done + self.assertEqual(self.reconciler.stats['pop_queue'], 1) + self.assertEqual(deleted_container_entries, + [('.misplaced_objects', '3600', '1:/AUTH_bob/c/o1')]) + self.assertEqual(self.reconciler.stats['success'], 1) + + def test_object_move_dest_object_older_than_queue_entry(self): + self._mock_listing({ + (None, "/.misplaced_objects/36000/1:/AUTH_bob/c/o1"): 36123.38393, + (1, "/AUTH_bob/c/o1"): 36123.38393, + (0, "/AUTH_bob/c/o1"): 36123.38393 - 0.00001, # slightly older + }) + self._mock_oldest_spi({'c': 0}) + deleted_container_entries = self._run_once() + + # we found a misplaced object + self.assertEqual(self.reconciler.stats['misplaced_object'], 1) + self.assertEqual( + self.fake_swift.calls, + [('GET', self.current_container_path), + ('GET', '/v1/.misplaced_objects' + listing_qs('')), + ('GET', '/v1/.misplaced_objects' + listing_qs('36000')), + ('GET', '/v1/.misplaced_objects/36000' + listing_qs('')), + ('GET', '/v1/.misplaced_objects/36000' + + listing_qs('1:/AUTH_bob/c/o1'))]) + # and since our version is *newer*, we overwrite + self.assertEqual(self.reconciler.stats['copy_attempt'], 1) + self.assertEqual(self.reconciler.stats['copy_success'], 1) + self.assertEqual( + self.fake_swift.storage_policy[1].calls, + [('GET', '/v1/AUTH_bob/c/o1'), # 2 + ('DELETE', '/v1/AUTH_bob/c/o1')]) # 4 + delete_headers = self.fake_swift.storage_policy[1].headers[1] + self.assertEqual( + self.fake_swift.storage_policy[0].calls, + [('HEAD', '/v1/AUTH_bob/c/o1'), # 1 + ('PUT', '/v1/AUTH_bob/c/o1')]) # 3 + # ... with a q_ts + offset 2 + put_headers = self.fake_swift.storage_policy[0].headers[1] + self.assertEqual(put_headers.get('X-Timestamp'), + normalize_timestamp(36123.38395)) + # then clean the dark matter + self.assertEqual(self.reconciler.stats['cleanup_attempt'], 1) + self.assertEqual(self.reconciler.stats['cleanup_success'], 1) + # ... with a q_ts + offset 1 + self.assertEqual(delete_headers.get('X-Timestamp'), + normalize_timestamp(36123.38394)) + + # and pop the queue + self.assertEqual(self.reconciler.stats['pop_queue'], 1) + self.assertEqual(deleted_container_entries, + [('.misplaced_objects', '36000', '1:/AUTH_bob/c/o1')]) + self.assertEqual(self.reconciler.stats['success'], 1) + + def test_object_move_put_fails(self): + # setup the cluster + self._mock_listing({ + (None, "/.misplaced_objects/36000/1:/AUTH_bob/c/o1"): 36123.383925, + (1, "/AUTH_bob/c/o1"): 36123.383925, + }) + self._mock_oldest_spi({'c': 0}) + + # make the put to dest fail! + self.fake_swift.storage_policy[0].register( + 'PUT', '/v1/AUTH_bob/c/o1', swob.HTTPServiceUnavailable, {}) + + # turn the crank + deleted_container_entries = self._run_once() + + # we find a misplaced object + self.assertEqual(self.reconciler.stats['misplaced_object'], 1) + self.assertEqual( + self.fake_swift.calls, + [('GET', self.current_container_path), + ('GET', '/v1/.misplaced_objects' + listing_qs('')), + ('GET', '/v1/.misplaced_objects' + listing_qs('36000')), + ('GET', '/v1/.misplaced_objects/36000' + listing_qs('')), + ('GET', '/v1/.misplaced_objects/36000' + + listing_qs('1:/AUTH_bob/c/o1'))]) + # and try to move it, but it fails + self.assertEqual(self.reconciler.stats['copy_attempt'], 1) + self.assertEqual( + self.fake_swift.storage_policy[1].calls, + [('GET', '/v1/AUTH_bob/c/o1')]) # 2 + self.assertEqual( + self.fake_swift.storage_policy[0].calls, + [('HEAD', '/v1/AUTH_bob/c/o1'), # 1 + ('PUT', '/v1/AUTH_bob/c/o1')]) # 3 + put_headers = self.fake_swift.storage_policy[0].headers[1] + # ...with q_ts + offset 2 (20-microseconds) + self.assertEqual(put_headers.get('X-Timestamp'), + normalize_timestamp(36123.383945)) + # but it failed + self.assertEqual(self.reconciler.stats['copy_success'], 0) + self.assertEqual(self.reconciler.stats['copy_failed'], 1) + # ... so we don't clean up the source + self.assertEqual(self.reconciler.stats['cleanup_attempt'], 0) + # and we don't pop the queue + self.assertEqual(deleted_container_entries, []) + self.assertEqual(self.reconciler.stats['unhandled_errors'], 0) + self.assertEqual(self.reconciler.stats['retry'], 1) + + def test_object_move_put_blows_up_crazy_town(self): + # setup the cluster + self._mock_listing({ + (None, "/.misplaced_objects/36000/1:/AUTH_bob/c/o1"): 36123.383925, + (1, "/AUTH_bob/c/o1"): 36123.383925, + }) + self._mock_oldest_spi({'c': 0}) + + # make the put to dest blow up crazy town + def blow_up(*args, **kwargs): + raise Exception('kaboom!') + + self.fake_swift.storage_policy[0].register( + 'PUT', '/v1/AUTH_bob/c/o1', blow_up, {}) + + # turn the crank + deleted_container_entries = self._run_once() + + # we find a misplaced object + self.assertEqual(self.reconciler.stats['misplaced_object'], 1) + self.assertEqual( + self.fake_swift.calls, + [('GET', self.current_container_path), + ('GET', '/v1/.misplaced_objects' + listing_qs('')), + ('GET', '/v1/.misplaced_objects' + listing_qs('36000')), + ('GET', '/v1/.misplaced_objects/36000' + listing_qs('')), + ('GET', '/v1/.misplaced_objects/36000' + + listing_qs('1:/AUTH_bob/c/o1'))]) + # and attempt to move it + self.assertEqual(self.reconciler.stats['copy_attempt'], 1) + self.assertEqual( + self.fake_swift.storage_policy[1].calls, + [('GET', '/v1/AUTH_bob/c/o1')]) # 2 + self.assertEqual( + self.fake_swift.storage_policy[0].calls, + [('HEAD', '/v1/AUTH_bob/c/o1'), # 1 + ('PUT', '/v1/AUTH_bob/c/o1')]) # 3 + put_headers = self.fake_swift.storage_policy[0].headers[1] + # ...with q_ts + offset 2 (20-microseconds) + self.assertEqual(put_headers.get('X-Timestamp'), + normalize_timestamp(36123.383945)) + # but it blows up hard + self.assertEqual(self.reconciler.stats['unhandled_error'], 1) + # so we don't cleanup + self.assertEqual(self.reconciler.stats['cleanup_attempt'], 0) + # and we don't pop the queue + self.assertEqual(self.reconciler.stats['pop_queue'], 0) + self.assertEqual(deleted_container_entries, []) + self.assertEqual(self.reconciler.stats['retry'], 1) + + def test_object_move_no_such_object_no_tombstone_recent(self): + q_ts = float(normalize_timestamp(time.time())) + container = str(int(q_ts // 3600 * 3600)) + q_path = '.misplaced_objects/%s' % container + + self._mock_listing({ + (None, "/%s/1:/AUTH_jeb/c/o1" % q_path): q_ts + }) + self._mock_oldest_spi({'c': 0}) + + deleted_container_entries = self._run_once() + + self.assertEqual( + self.fake_swift.calls, + [('GET', '/v1/.misplaced_objects/%s' % container + listing_qs('')), + ('GET', '/v1/.misplaced_objects/%s' % container + + listing_qs('1:/AUTH_jeb/c/o1')), + ('GET', '/v1/.misplaced_objects' + listing_qs('')), + ('GET', '/v1/.misplaced_objects' + listing_qs(container))]) + self.assertEqual( + self.fake_swift.storage_policy[0].calls, + [('HEAD', '/v1/AUTH_jeb/c/o1')], + ) + self.assertEqual( + self.fake_swift.storage_policy[1].calls, + [('GET', '/v1/AUTH_jeb/c/o1')], + ) + # the queue entry is recent enough that there could easily be + # tombstones on offline nodes or something, so we'll just leave it + # here and try again later + self.assertEqual(deleted_container_entries, []) + + def test_object_move_no_such_object_no_tombstone_ancient(self): + queue_ts = float(normalize_timestamp(time.time())) - \ + self.reconciler.reclaim_age * 1.1 + container = str(int(queue_ts // 3600 * 3600)) + + self._mock_listing({ + ( + None, "/.misplaced_objects/%s/1:/AUTH_jeb/c/o1" % container + ): queue_ts + }) + self._mock_oldest_spi({'c': 0}) + + deleted_container_entries = self._run_once() + + self.assertEqual( + self.fake_swift.calls, + [('GET', self.current_container_path), + ('GET', '/v1/.misplaced_objects' + listing_qs('')), + ('GET', '/v1/.misplaced_objects' + listing_qs(container)), + ('GET', '/v1/.misplaced_objects/%s' % container + listing_qs('')), + ('GET', '/v1/.misplaced_objects/%s' % container + + listing_qs('1:/AUTH_jeb/c/o1'))]) + self.assertEqual( + self.fake_swift.storage_policy[0].calls, + [('HEAD', '/v1/AUTH_jeb/c/o1')], + ) + self.assertEqual( + self.fake_swift.storage_policy[1].calls, + [('GET', '/v1/AUTH_jeb/c/o1')], + ) + + # the queue entry is old enough that the tombstones, if any, have + # probably been reaped, so we'll just give up + self.assertEqual( + deleted_container_entries, + [('.misplaced_objects', container, '1:/AUTH_jeb/c/o1')]) + + def test_delete_old_empty_queue_containers(self): + ts = float(normalize_timestamp(time.time())) - \ + self.reconciler.reclaim_age * 1.1 + container = str(int(ts // 3600 * 3600)) + older_ts = ts - 3600 + older_container = str(int(older_ts // 3600 * 3600)) + self._mock_listing({ + (None, "/.misplaced_objects/%s/" % container): 0, + (None, "/.misplaced_objects/%s/something" % older_container): 0, + }) + deleted_container_entries = self._run_once() + self.assertEqual(deleted_container_entries, []) + self.assertEqual( + self.fake_swift.calls, + [('GET', self.current_container_path), + ('GET', '/v1/.misplaced_objects' + listing_qs('')), + ('GET', '/v1/.misplaced_objects' + listing_qs(container)), + ('GET', '/v1/.misplaced_objects/%s' % container + listing_qs('')), + ('DELETE', '/v1/.misplaced_objects/%s' % container), + ('GET', '/v1/.misplaced_objects/%s' % older_container + + listing_qs('')), + ('GET', '/v1/.misplaced_objects/%s' % older_container + + listing_qs('something'))]) + self.assertEqual(self.reconciler.stats['invalid_record'], 1) + + def test_iter_over_old_containers_in_reverse(self): + step = reconciler.MISPLACED_OBJECTS_CONTAINER_DIVISOR + now = self.start_interval + containers = [] + for i in range(10): + container_ts = int(now - step * i) + container_name = str(container_ts // 3600 * 3600) + containers.append(container_name) + # add some old containers too + now -= self.reconciler.reclaim_age + old_containers = [] + for i in range(10): + container_ts = int(now - step * i) + container_name = str(container_ts // 3600 * 3600) + old_containers.append(container_name) + containers.sort() + old_containers.sort() + all_containers = old_containers + containers + self._mock_listing(dict(( + (None, "/.misplaced_objects/%s/" % container), 0 + ) for container in all_containers)) + deleted_container_entries = self._run_once() + self.assertEqual(deleted_container_entries, []) + last_container = all_containers[-1] + account_listing_calls = [ + ('GET', '/v1/.misplaced_objects' + listing_qs('')), + ('GET', '/v1/.misplaced_objects' + listing_qs(last_container)), + ] + new_container_calls = [ + ('GET', '/v1/.misplaced_objects/%s' % container + + listing_qs('')) for container in reversed(containers) + ][1:] # current_container get's skipped the second time around... + old_container_listings = [ + ('GET', '/v1/.misplaced_objects/%s' % container + + listing_qs('')) for container in reversed(old_containers) + ] + old_container_deletes = [ + ('DELETE', '/v1/.misplaced_objects/%s' % container) + for container in reversed(old_containers) + ] + old_container_calls = list(itertools.chain(*zip( + old_container_listings, old_container_deletes))) + self.assertEqual(self.fake_swift.calls, + [('GET', self.current_container_path)] + + account_listing_calls + new_container_calls + + old_container_calls) + + def test_error_in_iter_containers(self): + self._mock_listing({}) + + # make the listing return an error + self.fake_swift.storage_policy[None].register( + 'GET', '/v1/.misplaced_objects' + listing_qs(''), + swob.HTTPServiceUnavailable, {}) + + self._run_once() + self.assertEqual( + self.fake_swift.calls, + [('GET', self.current_container_path), + ('GET', '/v1/.misplaced_objects' + listing_qs(''))]) + self.assertEqual(self.reconciler.stats, {}) + errors = self.reconciler.logger.get_lines_for_level('error') + self.assertEqual(errors, [ + 'Error listing containers in account ' + '.misplaced_objects (Unexpected response: ' + '503 Service Unavailable)']) + + def test_unhandled_exception_in_reconcile(self): + self._mock_listing({}) + + # make the listing blow up + def blow_up(*args, **kwargs): + raise Exception('kaboom!') + + self.fake_swift.storage_policy[None].register( + 'GET', '/v1/.misplaced_objects' + listing_qs(''), + blow_up, {}) + self._run_once() + self.assertEqual( + self.fake_swift.calls, + [('GET', self.current_container_path), + ('GET', '/v1/.misplaced_objects' + listing_qs(''))]) + self.assertEqual(self.reconciler.stats, {}) + errors = self.reconciler.logger.get_lines_for_level('error') + self.assertEqual(errors, + ['Unhandled Exception trying to reconcile: ']) + + +if __name__ == '__main__': + unittest.main()