Adding a first stab at a general swift benchmark
This commit is contained in:
Chuck Thier
132
bin/swift-bench
Executable file
132
bin/swift-bench
Executable file
@@ -0,0 +1,132 @@
|
||||
#!/usr/bin/python
|
||||
# Copyright (c) 2010 OpenStack, LLC.
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
|
||||
# implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
import signal
|
||||
import uuid
|
||||
from optparse import OptionParser
|
||||
|
||||
from swift.common.bench import BenchController
|
||||
from swift.common.utils import readconf, NamedLogger
|
||||
|
||||
# The defaults should be sufficient to run swift-bench on a SAIO
|
||||
CONF_DEFAULTS = {
|
||||
'auth': '',
|
||||
'user': '',
|
||||
'key': '',
|
||||
'object_sources': '',
|
||||
'put_concurrency': '10',
|
||||
'get_concurrency': '10',
|
||||
'del_concurrency': '10',
|
||||
'concurrency': '',
|
||||
'object_size': '1',
|
||||
'num_objects': '1000',
|
||||
'num_gets': '10000',
|
||||
'delete': 'yes',
|
||||
'container_name': uuid.uuid4().hex,
|
||||
'use_proxy': 'yes',
|
||||
'url': '',
|
||||
'account': '',
|
||||
'devices': 'sdb1',
|
||||
'log_level': 'INFO',
|
||||
'timeout': '10',
|
||||
}
|
||||
|
||||
SAIO_DEFAULTS = {
|
||||
'auth': 'http://saio:11000/v1.0',
|
||||
'user': 'test:tester',
|
||||
'key': 'testing',
|
||||
}
|
||||
|
||||
if __name__ == '__main__':
|
||||
usage = "usage: %prog [OPTIONS] [CONF_FILE]"
|
||||
usage += """\n\nConf file with SAIO defaults:
|
||||
|
||||
[bench]
|
||||
auth = http://saio:11000/v1.0
|
||||
user = test:tester
|
||||
key = testing
|
||||
concurrency = 10
|
||||
object_size = 1
|
||||
num_objects = 1000
|
||||
num_gets = 10000
|
||||
delete = yes
|
||||
"""
|
||||
parser = OptionParser(usage=usage)
|
||||
parser.add_option('', '--saio', dest='saio', action='store_true',
|
||||
default=False, help='Run benchmark with SAIO defaults')
|
||||
parser.add_option('-A', '--auth', dest='auth',
|
||||
help='URL for obtaining an auth token')
|
||||
parser.add_option('-U', '--user', dest='user',
|
||||
help='User name for obtaining an auth token')
|
||||
parser.add_option('-K', '--key', dest='key',
|
||||
help='Key for obtaining an auth token')
|
||||
parser.add_option('-u', '--url', dest='url',
|
||||
help='Storage URL')
|
||||
parser.add_option('-c', '--concurrency', dest='concurrency',
|
||||
help='Number of concurrent connections to use')
|
||||
parser.add_option('-s', '--object-size', dest='object_size',
|
||||
help='Size of objects to PUT (in bytes)')
|
||||
parser.add_option('-n', '--num-objects', dest='num_objects',
|
||||
help='Number of objects to PUT')
|
||||
parser.add_option('-g', '--num-gets', dest='num_gets',
|
||||
help='Number of GET operations to perform')
|
||||
parser.add_option('-x', '--no-delete', dest='delete', action='store_false',
|
||||
help='If set, will not delete the objects created')
|
||||
|
||||
if len(sys.argv) == 1:
|
||||
parser.print_usage()
|
||||
sys.exit(1)
|
||||
options, args = parser.parse_args()
|
||||
if options.saio:
|
||||
CONF_DEFAULTS.update(SAIO_DEFAULTS)
|
||||
if args:
|
||||
conf = args[0]
|
||||
if not os.path.exists(conf):
|
||||
sys.exit("No such conf file: %s" % conf)
|
||||
conf = readconf(conf, 'bench', log_name='swift-bench',
|
||||
defaults=CONF_DEFAULTS)
|
||||
else:
|
||||
conf = CONF_DEFAULTS
|
||||
parser.set_defaults(**conf)
|
||||
options, _ = parser.parse_args()
|
||||
if options.concurrency is not '':
|
||||
options.put_concurrency = options.concurrency
|
||||
options.get_concurrency = options.concurrency
|
||||
options.del_concurrency = options.concurrency
|
||||
|
||||
def sigterm(signum, frame):
|
||||
sys.exit('Termination signal received.')
|
||||
signal.signal(signal.SIGTERM, sigterm)
|
||||
|
||||
logger = logging.getLogger()
|
||||
logger.setLevel({
|
||||
'debug': logging.DEBUG,
|
||||
'info': logging.INFO,
|
||||
'warning': logging.WARNING,
|
||||
'error': logging.ERROR,
|
||||
'critical': logging.CRITICAL}.get(
|
||||
options.log_level.lower(), logging.INFO))
|
||||
loghandler = logging.StreamHandler()
|
||||
logformat = logging.Formatter('%(asctime)s %(levelname)s %(message)s')
|
||||
loghandler.setFormatter(logformat)
|
||||
logger.addHandler(loghandler)
|
||||
logger = NamedLogger(logger, 'swift-bench')
|
||||
|
||||
controller = BenchController(logger, options)
|
||||
controller.run()
|
||||
3
setup.py
3
setup.py
@@ -74,7 +74,8 @@ setup(
|
||||
'bin/swift-object-server',
|
||||
'bin/swift-object-updater', 'bin/swift-proxy-server',
|
||||
'bin/swift-ring-builder', 'bin/swift-stats-populate',
|
||||
'bin/swift-stats-report'
|
||||
'bin/swift-stats-report',
|
||||
'bin/swift-bench',
|
||||
],
|
||||
entry_points={
|
||||
'paste.app_factory': [
|
||||
|
||||
236
swift/common/bench.py
Normal file
236
swift/common/bench.py
Normal file
@@ -0,0 +1,236 @@
|
||||
# Copyright (c) 2010 OpenStack, LLC.
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
|
||||
# implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
import uuid
|
||||
import time
|
||||
import random
|
||||
from urlparse import urlparse
|
||||
from contextlib import contextmanager
|
||||
|
||||
import eventlet.pools
|
||||
from eventlet.green.httplib import CannotSendRequest
|
||||
|
||||
from swift.common.utils import TRUE_VALUES
|
||||
from swift.common import client
|
||||
from swift.common import direct_client
|
||||
|
||||
|
||||
class ConnectionPool(eventlet.pools.Pool):
|
||||
|
||||
def __init__(self, url, size):
|
||||
self.url = url
|
||||
eventlet.pools.Pool.__init__(self, size, size)
|
||||
|
||||
def create(self):
|
||||
return client.http_connection(self.url)
|
||||
|
||||
|
||||
class Bench(object):
|
||||
|
||||
def __init__(self, logger, conf, names):
|
||||
self.logger = logger
|
||||
self.user = conf.user
|
||||
self.key = conf.key
|
||||
self.auth_url = conf.auth
|
||||
self.use_proxy = conf.use_proxy in TRUE_VALUES
|
||||
if self.use_proxy:
|
||||
url, token = client.get_auth(self.auth_url, self.user, self.key)
|
||||
self.token = token
|
||||
self.account = url.split('/')[-1]
|
||||
if conf.url == '':
|
||||
self.url = url
|
||||
else:
|
||||
self.url = conf.url
|
||||
else:
|
||||
self.token = 'SlapChop!'
|
||||
self.account = conf.account
|
||||
self.url = conf.url
|
||||
self.ip, self.port = self.url.split('/')[2].split(':')
|
||||
self.container_name = conf.container_name
|
||||
|
||||
self.object_size = int(conf.object_size)
|
||||
self.object_sources = conf.object_sources
|
||||
self.files = []
|
||||
if self.object_sources:
|
||||
self.object_sources = self.object_sources.split()
|
||||
self.files = [file(f, 'rb').read() for f in self.object_sources]
|
||||
|
||||
self.put_concurrency = int(conf.put_concurrency)
|
||||
self.get_concurrency = int(conf.get_concurrency)
|
||||
self.del_concurrency = int(conf.del_concurrency)
|
||||
self.total_objects = int(conf.num_objects)
|
||||
self.total_gets = int(conf.num_gets)
|
||||
self.timeout = int(conf.timeout)
|
||||
self.devices = conf.devices.split()
|
||||
self.names = names
|
||||
self.conn_pool = ConnectionPool(self.url,
|
||||
max(self.put_concurrency, self.get_concurrency,
|
||||
self.del_concurrency))
|
||||
|
||||
def _log_status(self, title):
|
||||
total = time.time() - self.beginbeat
|
||||
self.logger.info('%s %s [%s failures], %.01f/s' % (
|
||||
self.complete, title, self.failures,
|
||||
(float(self.complete) / total),
|
||||
))
|
||||
|
||||
@contextmanager
|
||||
def connection(self):
|
||||
try:
|
||||
hc = self.conn_pool.get()
|
||||
try:
|
||||
yield hc
|
||||
except CannotSendRequest:
|
||||
self.logger.info("CannotSendRequest. Skipping...")
|
||||
try:
|
||||
hc.close()
|
||||
except:
|
||||
pass
|
||||
self.failures += 1
|
||||
hc = self.conn_pool.create()
|
||||
finally:
|
||||
self.conn_pool.put(hc)
|
||||
|
||||
def run(self):
|
||||
pool = eventlet.GreenPool(self.concurrency)
|
||||
events = []
|
||||
self.beginbeat = self.heartbeat = time.time()
|
||||
self.heartbeat -= 13 # just to get the first report quicker
|
||||
self.failures = 0
|
||||
self.complete = 0
|
||||
for i in xrange(self.total):
|
||||
pool.spawn_n(self._run, i)
|
||||
pool.waitall()
|
||||
self._log_status(self.msg + ' **FINAL**')
|
||||
|
||||
def _run(self, thread):
|
||||
return
|
||||
|
||||
|
||||
class BenchController(object):
|
||||
|
||||
def __init__(self, logger, conf):
|
||||
self.logger = logger
|
||||
self.conf = conf
|
||||
self.names = []
|
||||
self.delete = conf.delete in TRUE_VALUES
|
||||
self.gets = int(conf.num_gets)
|
||||
|
||||
def run(self):
|
||||
puts = BenchPUT(self.logger, self.conf, self.names)
|
||||
puts.run()
|
||||
if self.gets:
|
||||
gets = BenchGET(self.logger, self.conf, self.names)
|
||||
gets.run()
|
||||
if self.delete:
|
||||
dels = BenchDELETE(self.logger, self.conf, self.names)
|
||||
dels.run()
|
||||
|
||||
|
||||
class BenchDELETE(Bench):
|
||||
|
||||
def __init__(self, logger, conf, names):
|
||||
Bench.__init__(self, logger, conf, names)
|
||||
self.concurrency = self.del_concurrency
|
||||
self.total = len(names)
|
||||
self.msg = 'DEL'
|
||||
|
||||
def _run(self, thread):
|
||||
if time.time() - self.heartbeat >= 15:
|
||||
self.heartbeat = time.time()
|
||||
self._log_status('DEL')
|
||||
device, partition, name = self.names.pop()
|
||||
with self.connection() as conn:
|
||||
try:
|
||||
if self.use_proxy:
|
||||
client.delete_object(self.url, self.token,
|
||||
self.container_name, name, http_conn=conn)
|
||||
else:
|
||||
node = {'ip': self.ip, 'port': self.port, 'device': device}
|
||||
direct_client.direct_delete_object(node, partition,
|
||||
self.account, self.container_name, name)
|
||||
except client.ClientException, e:
|
||||
self.logger.debug(str(e))
|
||||
self.failures += 1
|
||||
self.complete += 1
|
||||
|
||||
|
||||
class BenchGET(Bench):
|
||||
|
||||
def __init__(self, logger, conf, names):
|
||||
Bench.__init__(self, logger, conf, names)
|
||||
self.concurrency = self.get_concurrency
|
||||
self.total = self.total_gets
|
||||
self.msg = 'GETS'
|
||||
|
||||
def _run(self, thread):
|
||||
if time.time() - self.heartbeat >= 15:
|
||||
self.heartbeat = time.time()
|
||||
self._log_status('GETS')
|
||||
device, partition, name = random.choice(self.names)
|
||||
with self.connection() as conn:
|
||||
try:
|
||||
if self.use_proxy:
|
||||
client.get_object(self.url, self.token,
|
||||
self.container_name, name, http_conn=conn)
|
||||
else:
|
||||
node = {'ip': self.ip, 'port': self.port, 'device': device}
|
||||
direct_client.direct_get_object(node, partition,
|
||||
self.account, self.container_name, name)
|
||||
except client.ClientException, e:
|
||||
self.logger.debug(str(e))
|
||||
self.failures += 1
|
||||
self.complete += 1
|
||||
|
||||
|
||||
class BenchPUT(Bench):
|
||||
|
||||
def __init__(self, logger, conf, names):
|
||||
Bench.__init__(self, logger, conf, names)
|
||||
self.concurrency = self.put_concurrency
|
||||
self.total = self.total_objects
|
||||
self.msg = 'PUTS'
|
||||
if self.use_proxy:
|
||||
with self.connection() as conn:
|
||||
client.put_container(self.url, self.token,
|
||||
self.container_name, http_conn=conn)
|
||||
|
||||
def _run(self, thread):
|
||||
if time.time() - self.heartbeat >= 15:
|
||||
self.heartbeat = time.time()
|
||||
self._log_status('PUTS')
|
||||
name = uuid.uuid4().hex
|
||||
if self.object_sources:
|
||||
source = random.choice(self.files)
|
||||
else:
|
||||
source = '0' * self.object_size
|
||||
device = random.choice(self.devices)
|
||||
partition = str(random.randint(1, 3000))
|
||||
with self.connection() as conn:
|
||||
try:
|
||||
if self.use_proxy:
|
||||
client.put_object(self.url, self.token,
|
||||
self.container_name, name, source,
|
||||
content_length=len(source), http_conn=conn)
|
||||
else:
|
||||
node = {'ip': self.ip, 'port': self.port, 'device': device}
|
||||
direct_client.direct_put_object(node, partition,
|
||||
self.account, self.container_name, name, source,
|
||||
content_length=len(source))
|
||||
except client.ClientException, e:
|
||||
self.logger.debug(str(e))
|
||||
self.failures += 1
|
||||
self.names.append((device, partition, name))
|
||||
self.complete += 1
|
||||
@@ -18,7 +18,7 @@ Cloud Files client library used internally
|
||||
"""
|
||||
import socket
|
||||
from cStringIO import StringIO
|
||||
from httplib import HTTPConnection, HTTPException, HTTPSConnection
|
||||
from httplib import HTTPException, HTTPSConnection
|
||||
from re import compile, DOTALL
|
||||
from tokenize import generate_tokens, STRING, NAME, OP
|
||||
from urllib import quote as _quote, unquote
|
||||
@@ -29,6 +29,8 @@ try:
|
||||
except:
|
||||
from time import sleep
|
||||
|
||||
from swift.common.bufferedhttp \
|
||||
import BufferedHTTPConnection as HTTPConnection
|
||||
|
||||
def quote(value, safe='/'):
|
||||
"""
|
||||
|
||||
@@ -230,6 +230,62 @@ def direct_get_object(node, part, account, container, obj, conn_timeout=5,
|
||||
return resp_headers, object_body
|
||||
|
||||
|
||||
def direct_put_object(node, part, account, container, name, contents,
|
||||
content_length=None, etag=None, content_type=None,
|
||||
headers=None, conn_timeout=5, response_timeout=15,
|
||||
resp_chunk_size=None):
|
||||
"""
|
||||
Put object directly from the object server.
|
||||
|
||||
:param node: node dictionary from the ring
|
||||
:param part: partition the container is on
|
||||
:param account: account name
|
||||
:param container: container name
|
||||
:param name: object name
|
||||
:param contents: a string to read object data from
|
||||
:param content_length: value to send as content-length header
|
||||
:param etag: etag of contents
|
||||
:param content_type: value to send as content-type header
|
||||
:param headers: additional headers to include in the request
|
||||
:param conn_timeout: timeout in seconds for establishing the connection
|
||||
:param response_timeout: timeout in seconds for getting the response
|
||||
:param chunk_size: if defined, chunk size of data to send.
|
||||
:returns: etag from the server response
|
||||
"""
|
||||
# TODO: Add chunked puts
|
||||
path = '/%s/%s/%s' % (account, container, name)
|
||||
if headers is None:
|
||||
headers = {}
|
||||
if etag:
|
||||
headers['ETag'] = etag.strip('"')
|
||||
if content_length is not None:
|
||||
headers['Content-Length'] = str(content_length)
|
||||
if content_type is not None:
|
||||
headers['Content-Type'] = content_type
|
||||
else:
|
||||
headers['Content-Type'] = 'application/octet-stream'
|
||||
if not contents:
|
||||
headers['Content-Length'] = '0'
|
||||
headers['X-Timestamp'] = normalize_timestamp(time())
|
||||
with Timeout(conn_timeout):
|
||||
conn = http_connect(node['ip'], node['port'], node['device'], part,
|
||||
'PUT', path, headers=headers)
|
||||
conn.send(contents)
|
||||
with Timeout(response_timeout):
|
||||
resp = conn.getresponse()
|
||||
resp.read()
|
||||
if resp.status < 200 or resp.status >= 300:
|
||||
raise ClientException(
|
||||
'Object server %s:%s direct PUT %s gave status %s' %
|
||||
(node['ip'], node['port'],
|
||||
repr('/%s/%s%s' % (node['device'], part, path)),
|
||||
resp.status),
|
||||
http_host=node['ip'], http_port=node['port'],
|
||||
http_device=node['device'], http_status=resp.status,
|
||||
http_reason=resp.reason)
|
||||
return resp.getheader('etag').strip('"')
|
||||
|
||||
|
||||
def direct_delete_object(node, part, account, container, obj,
|
||||
conn_timeout=5, response_timeout=15, headers={}):
|
||||
"""
|
||||
|
||||
@@ -553,7 +553,7 @@ def cache_from_env(env):
|
||||
return item_from_env(env, 'swift.cache')
|
||||
|
||||
|
||||
def readconf(conf, section_name, log_name=None):
|
||||
def readconf(conf, section_name, log_name=None, defaults=None):
|
||||
"""
|
||||
Read config file and return config items as a dict
|
||||
|
||||
@@ -561,9 +561,12 @@ def readconf(conf, section_name, log_name=None):
|
||||
:param section_name: config section to read
|
||||
:param log_name: name to be used with logging (will use section_name if
|
||||
not defined)
|
||||
:param defaults: dict of default values to pre-populate the config with
|
||||
:returns: dict of config items
|
||||
"""
|
||||
c = ConfigParser()
|
||||
if defaults is None:
|
||||
defaults = {}
|
||||
c = ConfigParser(defaults)
|
||||
if not c.read(conf):
|
||||
print "Unable to read config file %s" % conf
|
||||
sys.exit(1)
|
||||
|
||||
Reference in New Issue
Block a user