openstack/zaqar
Code Issues Proposed changes

Split Control and Data planes of Storage layer

Browse Source 
To split Data/Control planes first separate groups for options are created.
Accordingly registered data and control plane options in the new group.
Modified drivers modules and many test cases.

Change-Id: I413b30b14ea605a7b3f22d6a51059a601af71e76
This commit is contained in:
Shaifali Agrawal 2014-11-17 03:27:29 -08:00
parent 57f0a87fc6
commit 44ceb3d392
15 changed files with 148 additions and 63 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
tests/etc/wsgi_mongodb.conf
View File

@ -10,9 +10,9 @@ storage = mongodb
[drivers:transport:wsgi]
port = 8888
[drivers:storage:mongodb]
[drivers:message_store:mongodb]
uri = mongodb://127.0.0.1:27017
database = zaqar_test
database = message_zaqar_test
max_reconnect_attempts = 3
reconnect_sleep = 0.001

6
tests/etc/wsgi_mongodb_pooled.conf
View File

@ -7,6 +7,10 @@ unreliable = True
transport = wsgi
storage = mongodb
[drivers:storage:mongodb]
[drivers:message_store:mongodb]
uri = mongodb://127.0.0.1:27017
database = zaqar_test
[drivers:management_store:mongodb]
uri = mongodb://127.0.0.1:27017
database = zaqar_test

2
tests/etc/wsgi_redis.conf
View File

@ -9,7 +9,7 @@ storage = redis
[drivers:transport:wsgi]
port = 8888
[drivers:storage:redis]
[drivers:message_store:redis]
uri = redis://127.0.0.1:6379
# NOTE(kgriffs): Reduce from the default of 10 to reduce the

7
tests/etc/wsgi_redis_pooled.conf
View File

@ -5,7 +5,12 @@ pooling = True
transport = wsgi
storage = redis
[drivers:storage:redis]
[drivers:message_store:redis]
uri = redis://127.0.0.1:6379
max_reconnect_attempts = 3
reconnect_sleep = 1
[drivers:management_store:redis]
uri = redis://127.0.0.1:6379
max_reconnect_attempts = 3
reconnect_sleep = 1

10
tests/unit/queues/storage/test_impl_mongodb.py
View File

@ -47,7 +47,9 @@ class MongodbSetupMixin(object):
self.driver.connection.drop_database(db)
def _prepare_conf(self):
self.config(options.MONGODB_GROUP,
self.config(options.MESSAGE_MONGODB_GROUP,
database=uuid.uuid4().hex)
self.config(options.MANAGEMENT_MONGODB_GROUP,
database=uuid.uuid4().hex)
@ -58,10 +60,10 @@ class MongodbUtilsTest(MongodbSetupMixin, testing.TestBase):
def setUp(self):
super(MongodbUtilsTest, self).setUp()
self.conf.register_opts(options.MONGODB_OPTIONS,
group=options.MONGODB_GROUP)
self.conf.register_opts(options.MESSAGE_MONGODB_OPTIONS,
group=options.MESSAGE_MONGODB_GROUP)
self.mongodb_conf = self.conf[options.MONGODB_GROUP]
self.mongodb_conf = self.conf[options.MESSAGE_MONGODB_GROUP]
MockDriver = collections.namedtuple('MockDriver', 'mongodb_conf')

6
tests/unit/queues/storage/test_impl_redis.py
View File

@ -63,10 +63,10 @@ class RedisUtilsTest(testing.TestBase):
def setUp(self):
super(RedisUtilsTest, self).setUp()
self.conf.register_opts(options.REDIS_OPTIONS,
group=options.REDIS_GROUP)
self.conf.register_opts(options.MESSAGE_REDIS_OPTIONS,
group=options.MESSAGE_REDIS_GROUP)
self.redis_conf = self.conf[options.REDIS_GROUP]
self.redis_conf = self.conf[options.MESSAGE_REDIS_GROUP]
MockDriver = collections.namedtuple('MockDriver', 'redis_conf')

8
zaqar/storage/mongodb/driver.py
View File

@ -70,7 +70,7 @@ class DataDriver(storage.DataDriverBase):
def __init__(self, conf, cache):
super(DataDriver, self).__init__(conf, cache)
self.mongodb_conf = self.conf[options.MONGODB_GROUP]
self.mongodb_conf = self.conf[options.MESSAGE_MONGODB_GROUP]
conn = self.connection
server_version = conn.server_info()['version']
@ -179,10 +179,10 @@ class ControlDriver(storage.ControlDriverBase):
def __init__(self, conf, cache):
super(ControlDriver, self).__init__(conf, cache)
self.conf.register_opts(options.MONGODB_OPTIONS,
group=options.MONGODB_GROUP)
self.conf.register_opts(options.MANAGEMENT_MONGODB_OPTIONS,
group=options.MANAGEMENT_MONGODB_GROUP)
self.mongodb_conf = self.conf[options.MONGODB_GROUP]
self.mongodb_conf = self.conf[options.MANAGEMENT_MONGODB_GROUP]
@decorators.lazy_property(write=False)
def connection(self):

108
zaqar/storage/mongodb/options.py
View File

@ -16,20 +16,31 @@
from oslo.config import cfg
_deprecated_group = 'drivers:storage:mongodb'
MONGODB_OPTIONS = (
# options that are common to both management and message storage
_COMMON_OPTIONS = (
cfg.StrOpt('ssl_keyfile',
deprecated_opts=[cfg.DeprecatedOpt(
'ssl_keyfile',
group=_deprecated_group), ],
help=('The private keyfile used to identify the local '
'connection against mongod. If included with the '
'``certifle`` then only the ``ssl_certfile`` '
'is needed.')),
cfg.StrOpt('ssl_certfile',
deprecated_opts=[cfg.DeprecatedOpt(
'ssl_certfile',
group=_deprecated_group), ],
help=('The certificate file used to identify the local '
'connection against mongod.')),
cfg.StrOpt('ssl_cert_reqs', default='CERT_REQUIRED',
deprecated_opts=[cfg.DeprecatedOpt(
'ssl_cert_reqs',
group=_deprecated_group), ],
help=('Specifies whether a certificate is required from '
'the other side of the connection, and whether it '
'will be validated if provided. It must be one of '
@ -41,20 +52,77 @@ MONGODB_OPTIONS = (
'must point to a file of CA certificates.')),
cfg.StrOpt('ssl_ca_certs',
deprecated_opts=[cfg.DeprecatedOpt(
'ssl_ca_certs',
group=_deprecated_group), ],
help=('The ca_certs file contains a set of concatenated '
'"certification authority" certificates, which are '
'used to validate certificates passed from the other '
'end of the connection.')),
cfg.StrOpt('uri',
deprecated_opts=[cfg.DeprecatedOpt(
'uri',
group=_deprecated_group), ],
help=('Mongodb Connection URI. If ssl connection enabled, '
'then ``ssl_keyfile``, ``ssl_certfile``, '
'``ssl_cert_reqs``, ``ssl_ca_certs`` need to be set '
'accordingly.')),
cfg.StrOpt('database', default='zaqar', help='Database name.'),
cfg.StrOpt('database', default='zaqar',
deprecated_opts=[cfg.DeprecatedOpt(
'database',
group=_deprecated_group), ],
help='Database name.'),
cfg.IntOpt('max_attempts', default=1000,
deprecated_opts=[cfg.DeprecatedOpt(
'max_attempts',
group=_deprecated_group), ],
help=('Maximum number of times to retry a failed operation. '
'Currently only used for retrying a message post.')),
cfg.FloatOpt('max_retry_sleep', default=0.1,
deprecated_opts=[cfg.DeprecatedOpt(
'max_retry_sleep',
group=_deprecated_group), ],
help=('Maximum sleep interval between retries '
'(actual sleep time increases linearly '
'according to number of attempts performed).')),
cfg.FloatOpt('max_retry_jitter', default=0.005,
deprecated_opts=[cfg.DeprecatedOpt(
'max_retry_jitter',
group=_deprecated_group), ],
help=('Maximum jitter interval, to be added to the '
'sleep interval, in order to decrease probability '
'that parallel requests will retry at the '
'same instant.')),
cfg.IntOpt('max_reconnect_attempts', default=10,
deprecated_opts=[cfg.DeprecatedOpt(
'max_reconnect_attempts',
group=_deprecated_group), ],
help=('Maximum number of times to retry an operation that '
'failed due to a primary node failover.')),
cfg.FloatOpt('reconnect_sleep', default=0.020,
deprecated_opts=[cfg.DeprecatedOpt(
'reconnect_sleep',
group=_deprecated_group), ],
help=('Base sleep interval between attempts to reconnect '
'after a primary node failover. '
'The actual sleep time increases exponentially (power '
'of 2) each time the operation is retried.')),
)
MANAGEMENT_MONGODB_OPTIONS = _COMMON_OPTIONS
MESSAGE_MONGODB_OPTIONS = _COMMON_OPTIONS + (
# options used only by message_storage
cfg.IntOpt('partitions', default=2,
deprecated_opts=[cfg.DeprecatedOpt(
'partitions',
group=_deprecated_group), ],
help=('Number of databases across which to '
'partition message data, in order to '
'reduce writer lock %. DO NOT change '
@ -63,35 +131,17 @@ MONGODB_OPTIONS = (
'should not need a large number of partitions '
'to improve performance, esp. if deploying '
'MongoDB on SSD storage.')),
cfg.IntOpt('max_attempts', default=1000,
help=('Maximum number of times to retry a failed operation. '
'Currently only used for retrying a message post.')),
cfg.FloatOpt('max_retry_sleep', default=0.1,
help=('Maximum sleep interval between retries '
'(actual sleep time increases linearly '
'according to number of attempts performed).')),
cfg.FloatOpt('max_retry_jitter', default=0.005,
help=('Maximum jitter interval, to be added to the '
'sleep interval, in order to decrease probability '
'that parallel requests will retry at the '
'same instant.')),
cfg.IntOpt('max_reconnect_attempts', default=10,
help=('Maximum number of times to retry an operation that '
'failed due to a primary node failover.')),
cfg.FloatOpt('reconnect_sleep', default=0.020,
help=('Base sleep interval between attempts to reconnect '
'after a primary node failover. '
'The actual sleep time increases exponentially (power '
'of 2) each time the operation is retried.')),
)
MONGODB_GROUP = 'drivers:storage:mongodb'
MANAGEMENT_MONGODB_GROUP = 'drivers:management_store:mongodb'
MESSAGE_MONGODB_GROUP = 'drivers:message_store:mongodb'
def _config_options():
return [(MONGODB_GROUP, MONGODB_OPTIONS)]
"""Used by config generators.
Returns a list of (group-name, oslo.config-options) tuples
for management and message storage.
"""
return [(MANAGEMENT_MONGODB_GROUP, MANAGEMENT_MONGODB_OPTIONS),
(MESSAGE_MONGODB_GROUP, MESSAGE_MONGODB_OPTIONS)]

8
zaqar/storage/redis/driver.py
View File

@ -144,7 +144,7 @@ class DataDriver(storage.DataDriverBase):
def __init__(self, conf, cache):
super(DataDriver, self).__init__(conf, cache)
self.redis_conf = self.conf[options.REDIS_GROUP]
self.redis_conf = self.conf[options.MESSAGE_REDIS_GROUP]
server_version = self.connection.info()['redis_version']
if tuple(map(int, server_version.split('.'))) < (2, 6):
@ -198,10 +198,10 @@ class ControlDriver(storage.ControlDriverBase):
def __init__(self, conf, cache):
super(ControlDriver, self).__init__(conf, cache)
self.conf.register_opts(options.REDIS_OPTIONS,
group=options.REDIS_GROUP)
self.conf.register_opts(options.MANAGEMENT_REDIS_OPTIONS,
group=options.MANAGEMENT_REDIS_GROUP)
self.redis_conf = self.conf[options.REDIS_GROUP]
self.redis_conf = self.conf[options.MANAGEMENT_REDIS_GROUP]
@decorators.lazy_property(write=False)
def connection(self):

23
zaqar/storage/redis/options.py
View File

@ -16,9 +16,13 @@
from oslo.config import cfg
REDIS_OPTIONS = (
_deprecated_group = 'drivers:storage:redis'
# options common to management and message storage
_COMMON_REDIS_OPTIONS = (
cfg.StrOpt('uri', default="redis://127.0.0.1:6379",
deprecated_opts=[cfg.DeprecatedOpt(
'uri',
group=_deprecated_group), ],
help=('Redis connection URI, taking one of three forms. '
'For a direct connection to a Redis server, use '
'the form "redis://host[:port][?options]", where '
@ -40,17 +44,28 @@ REDIS_OPTIONS = (
'"socket_timeout" defaults to 0.1 seconds.')),
cfg.IntOpt('max_reconnect_attempts', default=10,
deprecated_opts=[cfg.DeprecatedOpt(
'max_reconnect_attempts',
group=_deprecated_group), ],
help=('Maximum number of times to retry an operation that '
'failed due to a redis node failover.')),
cfg.FloatOpt('reconnect_sleep', default=1.0,
deprecated_opts=[cfg.DeprecatedOpt(
'reconnect_sleep',
group=_deprecated_group), ],
help=('Base sleep interval between attempts to reconnect '
'after a redis node failover. '))
)
REDIS_GROUP = 'drivers:storage:redis'
MANAGEMENT_REDIS_OPTIONS = _COMMON_REDIS_OPTIONS
MESSAGE_REDIS_OPTIONS = _COMMON_REDIS_OPTIONS
MANAGEMENT_REDIS_GROUP = 'drivers:management_store:redis'
MESSAGE_REDIS_GROUP = 'drivers:message_store:redis'
def _config_options():
return [(REDIS_GROUP, REDIS_OPTIONS)]
return [(MANAGEMENT_REDIS_GROUP, MANAGEMENT_REDIS_OPTIONS),
(MESSAGE_REDIS_GROUP, MESSAGE_REDIS_OPTIONS)]

8
zaqar/storage/sqlalchemy/driver.py
View File

@ -37,7 +37,7 @@ class DataDriver(storage.DataDriverBase):
def __init__(self, conf, cache):
super(DataDriver, self).__init__(conf, cache)
self.sqlalchemy_conf = self.conf[options.SQLALCHEMY_GROUP]
self.sqlalchemy_conf = self.conf[options.MESSAGE_SQLALCHEMY_GROUP]
LOG.warn(_('sqlalchemy\'s data plane driver will be removed during '
'the next release. Please, consider moving your data to '
'one of the other supported drivers.'))
@ -149,9 +149,9 @@ class ControlDriver(storage.ControlDriverBase):
def __init__(self, conf, cache):
super(ControlDriver, self).__init__(conf, cache)
self.conf.register_opts(options.SQLALCHEMY_OPTIONS,
group=options.SQLALCHEMY_GROUP)
self.sqlalchemy_conf = self.conf[options.SQLALCHEMY_GROUP]
self.conf.register_opts(options.MANAGEMENT_SQLALCHEMY_OPTIONS,
group=options.MANAGEMENT_SQLALCHEMY_GROUP)
self.sqlalchemy_conf = self.conf[options.MANAGEMENT_SQLALCHEMY_GROUP]
@decorators.lazy_property(write=False)
def engine(self, *args, **kwargs):

15
zaqar/storage/sqlalchemy/options.py
View File

@ -16,13 +16,22 @@
from oslo.config import cfg
SQLALCHEMY_OPTIONS = (
_deprecated_group = 'drivers:storage:sqlalchemy'
_COMMON_SQLALCHEMY_OPTIONS = (
cfg.StrOpt('uri', default='sqlite:///:memory:',
deprecated_opts=[cfg.DeprecatedOpt(
'uri',
group=_deprecated_group), ],
help='An sqlalchemy URL'),
)
SQLALCHEMY_GROUP = 'drivers:storage:sqlalchemy'
MANAGEMENT_SQLALCHEMY_OPTIONS = _COMMON_SQLALCHEMY_OPTIONS
MESSAGE_SQLALCHEMY_OPTIONS = _COMMON_SQLALCHEMY_OPTIONS
MANAGEMENT_SQLALCHEMY_GROUP = 'drivers:management_store:sqlalchemy'
MESSAGE_SQLALCHEMY_GROUP = 'drivers:message_store:sqlalchemy'
def _config_options():
return [(SQLALCHEMY_GROUP, SQLALCHEMY_OPTIONS)]
return [(MANAGEMENT_SQLALCHEMY_GROUP, MANAGEMENT_SQLALCHEMY_OPTIONS),
(MESSAGE_SQLALCHEMY_GROUP, MESSAGE_SQLALCHEMY_OPTIONS)]

2
zaqar/storage/utils.py
View File

@ -43,7 +43,7 @@ def dynamic_conf(uri, options, conf=None):
# NOTE(cpp-cabrera): parse storage-specific opts:
# 'drivers:storage:{type}'
storage_opts = utils.dict_to_conf({'uri': uri, 'options': options})
storage_group = u'drivers:storage:%s' % storage_type
storage_group = u'drivers:message_store:%s' % storage_type
# NOTE(cpp-cabrera): register those options!
if conf is None:

2
zaqar/tests/queues/transport/wsgi/v1/test_messages.py
View File

@ -37,7 +37,7 @@ class MessagesBaseTest(base.V1Base):
if self.conf.pooling:
for i in range(4):
uri = self.conf['drivers:storage:mongodb'].uri
uri = self.conf['drivers:management_store:mongodb'].uri
doc = {'weight': 100, 'uri': uri}
self.simulate_put(self.url_prefix + '/pools/' + str(i),
body=jsonutils.dumps(doc))

2
zaqar/tests/queues/transport/wsgi/v1_1/test_messages.py
View File

@ -39,7 +39,7 @@ class MessagesBaseTest(base.V1_1Base):
if self.conf.pooling:
for i in range(4):
uri = self.conf['drivers:storage:mongodb'].uri
uri = self.conf['drivers:management_store:mongodb'].uri
doc = {'weight': 100, 'uri': uri}
self.simulate_put(self.url_prefix + '/pools/' + str(i),
body=jsonutils.dumps(doc))