
We stopped writing pickled rings more than twelve years ago. Any cluster that was going to upgrade from then has, or can pick any of the multitude of intermediary releases to pause at and push rings. We can also safely assume that regions will be present for devices; that change is nearly as old. As a side-effect, clean up some old tests that did nonsense things like having 7 assignments per row for a part-power-2 ring. UpgradeImpact: remove ability to read rings written by swift <1.7.0 circa 2012 Related-Change: I799b9a4c894d54fb16592443904ac055b2638e2d Related-Change: Ifefbb839cdcf033e6c9201fadca95224c7303a29 Signed-off-by: Tim Burke <tim.burke@gmail.com> Change-Id: Ic8322b18d51b40f586cb217a0d1b2f345e1d8df6
340 lines
15 KiB
Python
340 lines
15 KiB
Python
# Copyright (c) 2010-2016 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.
|
|
|
|
"""
|
|
Provides helper functions for unit tests.
|
|
|
|
This cannot be in test/unit/__init__.py because that module is imported by the
|
|
py34 unit test job and there are imports here that end up importing modules
|
|
that are not yet ported to py34, such wsgi.py which import mimetools.
|
|
"""
|
|
import os
|
|
from tempfile import mkdtemp
|
|
import time
|
|
import warnings
|
|
|
|
from eventlet import spawn, wsgi
|
|
from unittest import mock
|
|
from shutil import rmtree
|
|
|
|
import swift
|
|
from swift.account import server as account_server
|
|
from swift.common import storage_policy
|
|
from swift.common.ring import RingData
|
|
from swift.common.storage_policy import StoragePolicy, ECStoragePolicy
|
|
from swift.common.middleware import listing_formats, proxy_logging
|
|
from swift.common import utils
|
|
from swift.common.utils import mkdirs, normalize_timestamp, NullLogger
|
|
from swift.common.http_protocol import SwiftHttpProtocol
|
|
from swift.container import server as container_server
|
|
from swift.obj import server as object_server
|
|
from swift.proxy import server as proxy_server
|
|
import swift.proxy.controllers.obj
|
|
|
|
from test import listen_zero
|
|
from test.debug_logger import debug_logger
|
|
from test.unit import write_fake_ring, DEFAULT_TEST_EC_TYPE, connect_tcp, \
|
|
readuntil2crlfs
|
|
|
|
|
|
def setup_servers(the_object_server=object_server, extra_conf=None):
|
|
"""
|
|
Setup proxy, account, container and object servers using a set of fake
|
|
rings and policies.
|
|
|
|
:param the_object_server: The object server module to use (optional,
|
|
defaults to swift.obj.server)
|
|
:param extra_conf: A dict of config options that will update the basic
|
|
config passed to all server instances.
|
|
:returns: A dict containing the following entries:
|
|
orig_POLICIES: the value of storage_policy.POLICIES prior to
|
|
it being patched with fake policies
|
|
orig_SysLogHandler: the value of utils.logs.SysLogHandler
|
|
prior to it being patched
|
|
testdir: root directory used for test files
|
|
test_POLICIES: a StoragePolicyCollection of fake policies
|
|
test_servers: a tuple of test server instances
|
|
test_sockets: a tuple of sockets used by test servers
|
|
test_coros: a tuple of greenthreads in which test servers are
|
|
running
|
|
"""
|
|
context = {
|
|
"orig_POLICIES": storage_policy._POLICIES,
|
|
"orig_SysLogHandler": utils.logs.SysLogHandler}
|
|
|
|
utils.HASH_PATH_SUFFIX = b'endcap'
|
|
utils.logs.SysLogHandler = mock.MagicMock()
|
|
# Since we're starting up a lot here, we're going to test more than
|
|
# just chunked puts; we're also going to test parts of
|
|
# proxy_server.Application we couldn't get to easily otherwise.
|
|
context["testdir"] = _testdir = \
|
|
os.path.join(mkdtemp(), 'tmp_test_proxy_server_chunked')
|
|
mkdirs(_testdir)
|
|
rmtree(_testdir)
|
|
for drive in ('sda1', 'sdb1', 'sdc1', 'sdd1', 'sde1',
|
|
'sdf1', 'sdg1', 'sdh1', 'sdi1', 'sdj1',
|
|
'sdk1', 'sdl1'):
|
|
mkdirs(os.path.join(_testdir, drive, 'tmp'))
|
|
conf = {'devices': _testdir, 'swift_dir': _testdir,
|
|
'mount_check': 'false', 'allowed_headers':
|
|
'content-encoding, x-object-manifest, content-disposition, foo',
|
|
'allow_versions': 't', 'node_timeout': 20}
|
|
if extra_conf:
|
|
conf.update(extra_conf)
|
|
context['conf'] = conf
|
|
prolis = listen_zero()
|
|
acc1lis = listen_zero()
|
|
acc2lis = listen_zero()
|
|
con1lis = listen_zero()
|
|
con2lis = listen_zero()
|
|
obj1lis = listen_zero()
|
|
obj2lis = listen_zero()
|
|
obj3lis = listen_zero()
|
|
obj4lis = listen_zero()
|
|
obj5lis = listen_zero()
|
|
obj6lis = listen_zero()
|
|
objsocks = [obj1lis, obj2lis, obj3lis, obj4lis, obj5lis, obj6lis]
|
|
context["test_sockets"] = \
|
|
(prolis, acc1lis, acc2lis, con1lis, con2lis, obj1lis, obj2lis, obj3lis,
|
|
obj4lis, obj5lis, obj6lis)
|
|
account_ring_path = os.path.join(_testdir, 'account.ring.gz')
|
|
account_devs = [
|
|
{'port': acc1lis.getsockname()[1]},
|
|
{'port': acc2lis.getsockname()[1]},
|
|
]
|
|
write_fake_ring(account_ring_path, *account_devs)
|
|
container_ring_path = os.path.join(_testdir, 'container.ring.gz')
|
|
container_devs = [
|
|
{'port': con1lis.getsockname()[1]},
|
|
{'port': con2lis.getsockname()[1]},
|
|
]
|
|
write_fake_ring(container_ring_path, *container_devs)
|
|
storage_policy._POLICIES = storage_policy.StoragePolicyCollection([
|
|
StoragePolicy(0, 'zero', True),
|
|
StoragePolicy(1, 'one', False),
|
|
StoragePolicy(2, 'two', False),
|
|
ECStoragePolicy(3, 'ec', ec_type=DEFAULT_TEST_EC_TYPE,
|
|
ec_ndata=2, ec_nparity=1, ec_segment_size=4096),
|
|
ECStoragePolicy(4, 'ec-dup', ec_type=DEFAULT_TEST_EC_TYPE,
|
|
ec_ndata=2, ec_nparity=1, ec_segment_size=4096,
|
|
ec_duplication_factor=2)])
|
|
obj_rings = {
|
|
0: ('sda1', 'sdb1'),
|
|
1: ('sdc1', 'sdd1'),
|
|
2: ('sde1', 'sdf1'),
|
|
# sdg1, sdh1, sdi1 taken by policy 3 (see below)
|
|
}
|
|
for policy_index, devices in obj_rings.items():
|
|
policy = storage_policy.POLICIES[policy_index]
|
|
obj_ring_path = os.path.join(_testdir, policy.ring_name + '.ring.gz')
|
|
obj_devs = [
|
|
{'port': objsock.getsockname()[1], 'device': dev}
|
|
for objsock, dev in zip(objsocks, devices)]
|
|
write_fake_ring(obj_ring_path, *obj_devs)
|
|
|
|
# write_fake_ring can't handle a 3-element ring, and the EC policy needs
|
|
# at least 6 devs to work with (ec_k=2, ec_m=1, duplication_factor=2),
|
|
# so we do it manually
|
|
devs = [{'id': 0, 'region': 1, 'zone': 0, 'device': 'sdg1',
|
|
'ip': '127.0.0.1', 'port': obj1lis.getsockname()[1]},
|
|
{'id': 1, 'region': 1, 'zone': 0, 'device': 'sdh1',
|
|
'ip': '127.0.0.1', 'port': obj2lis.getsockname()[1]},
|
|
{'id': 2, 'region': 1, 'zone': 0, 'device': 'sdi1',
|
|
'ip': '127.0.0.1', 'port': obj3lis.getsockname()[1]},
|
|
{'id': 3, 'region': 1, 'zone': 0, 'device': 'sdj1',
|
|
'ip': '127.0.0.1', 'port': obj4lis.getsockname()[1]},
|
|
{'id': 4, 'region': 1, 'zone': 0, 'device': 'sdk1',
|
|
'ip': '127.0.0.1', 'port': obj5lis.getsockname()[1]},
|
|
{'id': 5, 'region': 1, 'zone': 0, 'device': 'sdl1',
|
|
'ip': '127.0.0.1', 'port': obj6lis.getsockname()[1]}]
|
|
pol3_replica2part2dev_id = [[0, 1, 2, 0],
|
|
[1, 2, 0, 1],
|
|
[2, 0, 1, 2]]
|
|
pol4_replica2part2dev_id = [[0, 1, 2, 3],
|
|
[1, 2, 3, 4],
|
|
[2, 3, 4, 5],
|
|
[3, 4, 5, 0],
|
|
[4, 5, 0, 1],
|
|
[5, 0, 1, 2]]
|
|
obj3_ring_path = os.path.join(
|
|
_testdir, storage_policy.POLICIES[3].ring_name + '.ring.gz')
|
|
part_shift = 30
|
|
RingData(pol3_replica2part2dev_id, devs, part_shift).save(obj3_ring_path)
|
|
|
|
obj4_ring_path = os.path.join(
|
|
_testdir, storage_policy.POLICIES[4].ring_name + '.ring.gz')
|
|
part_shift = 30
|
|
RingData(pol4_replica2part2dev_id, devs, part_shift).save(obj4_ring_path)
|
|
|
|
prosrv = proxy_server.Application(conf, logger=debug_logger('proxy'))
|
|
for policy in storage_policy.POLICIES:
|
|
# make sure all the rings are loaded
|
|
prosrv.get_object_ring(policy.idx)
|
|
# don't lose this one!
|
|
context["test_POLICIES"] = storage_policy._POLICIES
|
|
acc1srv = account_server.AccountController(
|
|
conf, logger=debug_logger('acct1'))
|
|
acc2srv = account_server.AccountController(
|
|
conf, logger=debug_logger('acct2'))
|
|
con1srv = container_server.ContainerController(
|
|
conf, logger=debug_logger('cont1'))
|
|
con2srv = container_server.ContainerController(
|
|
conf, logger=debug_logger('cont2'))
|
|
obj1srv = the_object_server.ObjectController(
|
|
conf, logger=debug_logger('obj1'))
|
|
obj2srv = the_object_server.ObjectController(
|
|
conf, logger=debug_logger('obj2'))
|
|
obj3srv = the_object_server.ObjectController(
|
|
conf, logger=debug_logger('obj3'))
|
|
obj4srv = the_object_server.ObjectController(
|
|
conf, logger=debug_logger('obj4'))
|
|
obj5srv = the_object_server.ObjectController(
|
|
conf, logger=debug_logger('obj5'))
|
|
obj6srv = the_object_server.ObjectController(
|
|
conf, logger=debug_logger('obj6'))
|
|
context["test_servers"] = \
|
|
(prosrv, acc1srv, acc2srv, con1srv, con2srv, obj1srv, obj2srv, obj3srv,
|
|
obj4srv, obj5srv, obj6srv)
|
|
nl = NullLogger()
|
|
logging_prosv = proxy_logging.ProxyLoggingMiddleware(
|
|
listing_formats.ListingFilter(prosrv, {}, logger=prosrv.logger),
|
|
conf, logger=prosrv.logger)
|
|
# Yes, eventlet, we know -- we have to support bad clients, though
|
|
warnings.filterwarnings(
|
|
'ignore', module='eventlet',
|
|
message='capitalize_response_headers is disabled')
|
|
prospa = spawn(wsgi.server, prolis, logging_prosv, nl,
|
|
protocol=SwiftHttpProtocol,
|
|
capitalize_response_headers=False)
|
|
acc1spa = spawn(wsgi.server, acc1lis, acc1srv, nl,
|
|
protocol=SwiftHttpProtocol,
|
|
capitalize_response_headers=False)
|
|
acc2spa = spawn(wsgi.server, acc2lis, acc2srv, nl,
|
|
protocol=SwiftHttpProtocol,
|
|
capitalize_response_headers=False)
|
|
con1spa = spawn(wsgi.server, con1lis, con1srv, nl,
|
|
protocol=SwiftHttpProtocol,
|
|
capitalize_response_headers=False)
|
|
con2spa = spawn(wsgi.server, con2lis, con2srv, nl,
|
|
protocol=SwiftHttpProtocol,
|
|
capitalize_response_headers=False)
|
|
obj1spa = spawn(wsgi.server, obj1lis, obj1srv, nl,
|
|
protocol=SwiftHttpProtocol,
|
|
capitalize_response_headers=False)
|
|
obj2spa = spawn(wsgi.server, obj2lis, obj2srv, nl,
|
|
protocol=SwiftHttpProtocol,
|
|
capitalize_response_headers=False)
|
|
obj3spa = spawn(wsgi.server, obj3lis, obj3srv, nl,
|
|
protocol=SwiftHttpProtocol,
|
|
capitalize_response_headers=False)
|
|
obj4spa = spawn(wsgi.server, obj4lis, obj4srv, nl,
|
|
protocol=SwiftHttpProtocol,
|
|
capitalize_response_headers=False)
|
|
obj5spa = spawn(wsgi.server, obj5lis, obj5srv, nl,
|
|
protocol=SwiftHttpProtocol,
|
|
capitalize_response_headers=False)
|
|
obj6spa = spawn(wsgi.server, obj6lis, obj6srv, nl,
|
|
protocol=SwiftHttpProtocol,
|
|
capitalize_response_headers=False)
|
|
context["test_coros"] = \
|
|
(prospa, acc1spa, acc2spa, con1spa, con2spa, obj1spa, obj2spa, obj3spa,
|
|
obj4spa, obj5spa, obj6spa)
|
|
# Create account
|
|
ts = normalize_timestamp(time.time())
|
|
partition, nodes = prosrv.account_ring.get_nodes('a')
|
|
for node in nodes:
|
|
conn = swift.proxy.controllers.obj.http_connect(node['ip'],
|
|
node['port'],
|
|
node['device'],
|
|
partition, 'PUT', '/a',
|
|
{'X-Timestamp': ts,
|
|
'x-trans-id': 'test'})
|
|
resp = conn.getresponse()
|
|
assert resp.status == 201
|
|
# Create another account
|
|
# used for account-to-account tests
|
|
ts = normalize_timestamp(time.time())
|
|
partition, nodes = prosrv.account_ring.get_nodes('a1')
|
|
for node in nodes:
|
|
conn = swift.proxy.controllers.obj.http_connect(node['ip'],
|
|
node['port'],
|
|
node['device'],
|
|
partition, 'PUT',
|
|
'/a1',
|
|
{'X-Timestamp': ts,
|
|
'x-trans-id': 'test'})
|
|
resp = conn.getresponse()
|
|
assert resp.status == 201
|
|
# Create containers, 1 per test policy
|
|
sock = connect_tcp(('localhost', prolis.getsockname()[1]))
|
|
fd = sock.makefile('rwb')
|
|
fd.write(b'PUT /v1/a/c HTTP/1.1\r\nHost: localhost\r\n'
|
|
b'Connection: close\r\nX-Auth-Token: t\r\n'
|
|
b'Content-Length: 0\r\n\r\n')
|
|
fd.flush()
|
|
headers = readuntil2crlfs(fd)
|
|
exp = b'HTTP/1.1 201'
|
|
assert headers[:len(exp)] == exp, "Expected '%s', encountered '%s'" % (
|
|
exp, headers[:len(exp)])
|
|
# Create container in other account
|
|
# used for account-to-account tests
|
|
sock = connect_tcp(('localhost', prolis.getsockname()[1]))
|
|
fd = sock.makefile('rwb')
|
|
fd.write(b'PUT /v1/a1/c1 HTTP/1.1\r\nHost: localhost\r\n'
|
|
b'Connection: close\r\nX-Auth-Token: t\r\n'
|
|
b'Content-Length: 0\r\n\r\n')
|
|
fd.flush()
|
|
headers = readuntil2crlfs(fd)
|
|
exp = b'HTTP/1.1 201'
|
|
assert headers[:len(exp)] == exp, "Expected '%s', encountered '%s'" % (
|
|
exp, headers[:len(exp)])
|
|
|
|
sock = connect_tcp(('localhost', prolis.getsockname()[1]))
|
|
fd = sock.makefile('rwb')
|
|
fd.write(
|
|
b'PUT /v1/a/c1 HTTP/1.1\r\nHost: localhost\r\n'
|
|
b'Connection: close\r\nX-Auth-Token: t\r\nX-Storage-Policy: one\r\n'
|
|
b'Content-Length: 0\r\n\r\n')
|
|
fd.flush()
|
|
headers = readuntil2crlfs(fd)
|
|
exp = b'HTTP/1.1 201'
|
|
assert headers[:len(exp)] == exp, \
|
|
"Expected %r, encountered %r" % (exp, headers[:len(exp)])
|
|
|
|
sock = connect_tcp(('localhost', prolis.getsockname()[1]))
|
|
fd = sock.makefile('rwb')
|
|
fd.write(
|
|
b'PUT /v1/a/c2 HTTP/1.1\r\nHost: localhost\r\n'
|
|
b'Connection: close\r\nX-Auth-Token: t\r\nX-Storage-Policy: two\r\n'
|
|
b'Content-Length: 0\r\n\r\n')
|
|
fd.flush()
|
|
headers = readuntil2crlfs(fd)
|
|
exp = b'HTTP/1.1 201'
|
|
assert headers[:len(exp)] == exp, \
|
|
"Expected '%s', encountered '%s'" % (exp, headers[:len(exp)])
|
|
return context
|
|
|
|
|
|
def teardown_servers(context):
|
|
for server in context["test_coros"]:
|
|
server.kill()
|
|
# We didn't start the proxy w/ run_server, so we have to kill the
|
|
# watchdog ourselves
|
|
context["test_servers"][0].watchdog.kill()
|
|
assert context["test_servers"][0].watchdog._run_gth is None
|
|
rmtree(os.path.dirname(context["testdir"]))
|
|
utils.logs.SysLogHandler = context["orig_SysLogHandler"]
|
|
storage_policy._POLICIES = context["orig_POLICIES"]
|