![Tim Burke](/assets/img/avatar_default.png)
This change introduces the concept of a "versioning mode" for versioned_writes. The following modes are supported: * stack When deleting, check whether any previous versions exist in the versions container. If none is found, the object is deleted. If the most-recent version in the versions container is not a delete marker, it is copied into the versioned container (overwriting the current version if one exists) and then deleted from the versions container. This preserves the previous behavior. If the most-recent version in the versions container is a delete marker and a current version exists in the versioned container, the current version is deleted. If the most-recent version in the versions container is a delete marker and no current version exists in the versioned container, we copy the next-most-recent version from the versions container into the versioned container (assuming it exists and is not a delete marker) and delete both the most-recent version (i.e., the delete marker) and the just-copied next-most-recent version from the versions container. With this mode, DELETEs to versioned containers "undo" operations on containers. Previously this was limited to undoing PUTs, but now it will also undo DELETEs performed while in "history" mode. * history When deleting, check whether a current version exists in the versioned container. If one is found, it is copied to the versions container. Then an empty "delete marker" object is also put into the versions container; this records when the object was deleted. Finally, the original current version is deleted from the versioned container. As a result, subsequent GETs or HEADs will return a 404, and container listings for the versioned container do not include the object. With this mode, DELETEs to versioned containers behave like DELETEs to other containers, but with a history of what has happened. Clients may specify (via a new X-Versions-Mode header) which mode a container should use. By default, the existing "stack" mode is used. Upgrade consideration: ====================== Clients should not use the "history" mode until all proxies in the cluster have been upgraded. Attempting to use the "history" mode during a rolling upgrade may result in some requests being served by proxies running old code (which necessarily uses the "stack" mode), leading to data loss. Change-Id: I555dc17fefd0aa9ade681aa156da24e018ebe74b
213 lines
7.6 KiB
Python
213 lines
7.6 KiB
Python
# Copyright (c) 2013 OpenStack Foundation
|
|
#
|
|
# 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.
|
|
|
|
# This stuff can't live in test/unit/__init__.py due to its swob dependency.
|
|
|
|
from collections import defaultdict, namedtuple
|
|
from hashlib import md5
|
|
from swift.common import swob
|
|
from swift.common.header_key_dict import HeaderKeyDict
|
|
from swift.common.request_helpers import is_user_meta, \
|
|
is_object_transient_sysmeta
|
|
from swift.common.swob import HTTPNotImplemented
|
|
from swift.common.utils import split_path
|
|
|
|
from test.unit import FakeLogger, FakeRing
|
|
|
|
|
|
class LeakTrackingIter(object):
|
|
def __init__(self, inner_iter, fake_swift, path):
|
|
self.inner_iter = inner_iter
|
|
self.fake_swift = fake_swift
|
|
self.path = path
|
|
|
|
def __iter__(self):
|
|
for x in self.inner_iter:
|
|
yield x
|
|
|
|
def close(self):
|
|
self.fake_swift.mark_closed(self.path)
|
|
|
|
|
|
FakeSwiftCall = namedtuple('FakeSwiftCall', ['method', 'path', 'headers'])
|
|
|
|
|
|
class FakeSwift(object):
|
|
"""
|
|
A good-enough fake Swift proxy server to use in testing middleware.
|
|
"""
|
|
ALLOWED_METHODS = [
|
|
'PUT', 'POST', 'DELETE', 'GET', 'HEAD', 'OPTIONS', 'REPLICATE']
|
|
|
|
def __init__(self):
|
|
self._calls = []
|
|
self._unclosed_req_paths = defaultdict(int)
|
|
self.req_method_paths = []
|
|
self.swift_sources = []
|
|
self.txn_ids = []
|
|
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 _find_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']
|
|
if method not in self.ALLOWED_METHODS:
|
|
raise HTTPNotImplemented()
|
|
|
|
path = env['PATH_INFO']
|
|
_, acc, cont, obj = split_path(env['PATH_INFO'], 0, 4,
|
|
rest_with_last=True)
|
|
if env.get('QUERY_STRING'):
|
|
path += '?' + env['QUERY_STRING']
|
|
|
|
if 'swift.authorize' in env:
|
|
resp = env['swift.authorize'](swob.Request(env))
|
|
if resp:
|
|
return resp(env, start_response)
|
|
|
|
req = swob.Request(env)
|
|
self.swift_sources.append(env.get('swift.source'))
|
|
self.txn_ids.append(env.get('swift.trans_id'))
|
|
|
|
try:
|
|
resp_class, raw_headers, body = self._find_response(method, path)
|
|
headers = HeaderKeyDict(raw_headers)
|
|
except KeyError:
|
|
if (env.get('QUERY_STRING')
|
|
and (method, env['PATH_INFO']) in self._responses):
|
|
resp_class, raw_headers, body = self._find_response(
|
|
method, env['PATH_INFO'])
|
|
headers = HeaderKeyDict(raw_headers)
|
|
elif method == 'HEAD' and ('GET', path) in self._responses:
|
|
resp_class, raw_headers, body = self._find_response(
|
|
'GET', path)
|
|
body = None
|
|
headers = HeaderKeyDict(raw_headers)
|
|
elif method == 'GET' and obj and path in self.uploaded:
|
|
resp_class = swob.HTTPOk
|
|
headers, body = self.uploaded[path]
|
|
else:
|
|
raise KeyError("Didn't find %r in allowed responses" % (
|
|
(method, path),))
|
|
|
|
# simulate object PUT
|
|
if method == 'PUT' and obj:
|
|
put_body = ''.join(iter(env['wsgi.input'].read, ''))
|
|
if 'swift.callback.update_footers' in env:
|
|
footers = HeaderKeyDict()
|
|
env['swift.callback.update_footers'](footers)
|
|
req.headers.update(footers)
|
|
etag = md5(put_body).hexdigest()
|
|
headers.setdefault('Etag', etag)
|
|
headers.setdefault('Content-Length', len(put_body))
|
|
|
|
# keep it for subsequent GET requests later
|
|
self.uploaded[path] = (dict(req.headers), put_body)
|
|
if "CONTENT_TYPE" in env:
|
|
self.uploaded[path][0]['Content-Type'] = env["CONTENT_TYPE"]
|
|
|
|
# simulate object POST
|
|
elif method == 'POST' and obj:
|
|
metadata, data = self.uploaded.get(path, ({}, None))
|
|
# select items to keep from existing...
|
|
new_metadata = dict(
|
|
(k, v) for k, v in metadata.items()
|
|
if (not is_user_meta('object', k) and not
|
|
is_object_transient_sysmeta(k)))
|
|
# apply from new
|
|
new_metadata.update(
|
|
dict((k, v) for k, v in req.headers.items()
|
|
if (is_user_meta('object', k) or
|
|
is_object_transient_sysmeta(k) or
|
|
k.lower == 'content-type')))
|
|
self.uploaded[path] = new_metadata, data
|
|
|
|
# note: tests may assume this copy of req_headers is case insensitive
|
|
# so we deliberately use a HeaderKeyDict
|
|
self._calls.append(
|
|
FakeSwiftCall(method, path, HeaderKeyDict(req.headers)))
|
|
|
|
# range requests ought to work, hence conditional_response=True
|
|
if isinstance(body, list):
|
|
resp = resp_class(
|
|
req=req, headers=headers, app_iter=body,
|
|
conditional_response=req.method in ('GET', 'HEAD'))
|
|
else:
|
|
resp = resp_class(
|
|
req=req, headers=headers, body=body,
|
|
conditional_response=req.method in ('GET', 'HEAD'))
|
|
wsgi_iter = resp(env, start_response)
|
|
self.mark_opened(path)
|
|
return LeakTrackingIter(wsgi_iter, self, path)
|
|
|
|
def mark_opened(self, path):
|
|
self._unclosed_req_paths[path] += 1
|
|
|
|
def mark_closed(self, path):
|
|
self._unclosed_req_paths[path] -= 1
|
|
|
|
@property
|
|
def unclosed_requests(self):
|
|
return {path: count
|
|
for path, count in self._unclosed_req_paths.items()
|
|
if count > 0}
|
|
|
|
@property
|
|
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
|
|
|
|
@property
|
|
def call_count(self):
|
|
return len(self._calls)
|
|
|
|
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)
|
|
|
|
|
|
class FakeAppThatExcepts(object):
|
|
MESSAGE = "We take exception to that!"
|
|
|
|
def __init__(self, exception_class=Exception):
|
|
self.exception_class = exception_class
|
|
|
|
def __call__(self, env, start_response):
|
|
raise self.exception_class(self.MESSAGE)
|