Implement mechanism to provide non-expiring keys in KVS
This patchset implements the ability to define non-expiring keys for dogpile.cache backends. The non-expiring keys are relevant in the case of drivers that can automatically remove keys after a given time (e.g. memcache). This new non-expiring-key functionality is currently only implemented for the provided memcached backend. bp: dogpile-kvs-backends Change-Id: I7e25e0049e5b8697c5cb67272b660519c3c3305e
This commit is contained in:
Morgan Fainberg
parent
359ef5345b
commit
57d02590f9
|
|
@ -465,6 +465,50 @@ KeyValueStore object named "TestKVSRegion" and a specific Memcached driver:
|
|||
kvs_store = kvs.get_key_value_store('TestKVSRegion')
|
||||
kvs_store.configure('openstack.kvs.Memcached', memcached_backend='Memcached')
|
||||
|
||||
The memcached backend supports a mechanism to supply an explicit TTL (in seconds) to all keys
|
||||
set via the KVS object. This is accomplished by passing the argument ``memcached_expire_time``
|
||||
as a keyword argument to the ``configure`` method. Passing the ``memcache_expire_time`` argument
|
||||
will cause the ``time`` argument to be added to all ``set`` and ``set_multi`` calls performed by
|
||||
the memcached client. ``memcached_expire_time`` is an argument exclusive to the memcached dogpile
|
||||
backend, and will be ignored if passed to another backend:
|
||||
|
||||
.. code:: python
|
||||
|
||||
kvs_store.configure('openstack.kvs.Memcached', memcached_backend='Memcached',
|
||||
memcached_expire_time=86400)
|
||||
|
||||
If an explicit TTL is configured via the ``memcached_expire_time`` argument, it is possible to
|
||||
exempt specific keys from receiving the TTL by passing the argument ``no_expiry_keys`` (list)
|
||||
as a keyword argument to the ``configure`` method. ``no_expiry_keys`` should be supported by
|
||||
all OpenStack-specific dogpile backends (memcached) that have the ability to set an explicit TTL:
|
||||
|
||||
.. code:: python
|
||||
|
||||
kvs_store.configure('openstack.kvs.Memcached', memcached_backend='Memcached',
|
||||
memcached_expire_time=86400, no_expiry_keys=['key', 'second_key', ...])
|
||||
|
||||
|
||||
.. NOTE::
|
||||
For the non-expiring keys functionality to work, the backend must support the ability for
|
||||
the region to set the key_mangler on it and have the attribute ``raw_no_expiry_keys``.
|
||||
In most cases, support for setting the key_mangler on the backend is handled by allowing
|
||||
the region object to set the ``key_mangler`` attribute on the backend.
|
||||
|
||||
The ``raw_no_expiry_keys`` attribute is expected to be used to hold the values of the
|
||||
keyword argument ``no_expiry_keys`` prior to hashing. It is the responsibility of the
|
||||
backend to use these raw values to determine if a key should be exempt from expiring
|
||||
and not set the TTL on the non-expiring keys when the ``set`` or ``set_multi`` methods are
|
||||
called.
|
||||
|
||||
Typically the key will be hashed by the region using its key_mangler method
|
||||
before being passed to the backend to set the value in the KeyValueStore. This
|
||||
means that in most cases, the backend will need to either pre-compute the hashed versions
|
||||
of the keys (when the key_mangler is set) and store a cached copy, or hash each item in
|
||||
the ``raw_no_expiry_keys`` attribute on each call to ``.set()`` and ``.set_multi()``. The
|
||||
``memcached`` backend handles this hashing and caching of the keys by utilizing an
|
||||
``@property`` method for the ``.key_mangler`` attribute on the backend and utilizing the
|
||||
associated ``.settr()`` method to front-load the hashing work at attribute set time.
|
||||
|
||||
Once a KVS object has been instantiated the method of interacting is the same as most memcache
|
||||
implementations:
|
||||
|
||||
|
|
|
|||
|
|
@ -83,6 +83,10 @@ class MemcachedBackend(manager.Manager):
|
|||
time `memcached`, `bmemcached`, `pylibmc` are valid).
|
||||
"""
|
||||
def __init__(self, arguments):
|
||||
self._key_mangler = None
|
||||
self.raw_no_expiry_keys = set(arguments.pop('no_expiry_keys', set()))
|
||||
self.no_expiry_hashed_keys = set()
|
||||
|
||||
self.lock_timeout = arguments.pop('lock_timeout', None)
|
||||
self.max_lock_attempts = arguments.pop('max_lock_attempts', 15)
|
||||
# NOTE(morganfainberg): Remove distributed locking from the arguments
|
||||
|
|
@ -110,6 +114,45 @@ class MemcachedBackend(manager.Manager):
|
|||
else:
|
||||
self.driver = VALID_DOGPILE_BACKENDS[backend](arguments)
|
||||
|
||||
def _get_set_arguments_driver_attr(self, exclude_expiry=False):
|
||||
|
||||
# NOTE(morganfainberg): Shallow copy the .set_arguments dict to
|
||||
# ensure no changes cause the values to change in the instance
|
||||
# variable.
|
||||
set_arguments = getattr(self.driver, 'set_arguments', {}).copy()
|
||||
|
||||
if exclude_expiry:
|
||||
# NOTE(morganfainberg): Explicitly strip out the 'time' key/value
|
||||
# from the set_arguments in the case that this key isn't meant
|
||||
# to expire
|
||||
set_arguments.pop('time', None)
|
||||
return set_arguments
|
||||
|
||||
def set(self, key, value):
|
||||
mapping = {key: value}
|
||||
self.set_multi(mapping)
|
||||
|
||||
def set_multi(self, mapping):
|
||||
mapping_keys = set(mapping.keys())
|
||||
no_expiry_keys = mapping_keys.intersection(self.no_expiry_hashed_keys)
|
||||
has_expiry_keys = mapping_keys.difference(self.no_expiry_hashed_keys)
|
||||
|
||||
if no_expiry_keys:
|
||||
# NOTE(morganfainberg): For keys that have expiry excluded,
|
||||
# bypass the backend and directly call the client. Bypass directly
|
||||
# to the client is required as the 'set_arguments' are applied to
|
||||
# all ``set`` and ``set_multi`` calls by the driver, by calling
|
||||
# the client directly it is possible to exclude the ``time``
|
||||
# argument to the memcached server.
|
||||
new_mapping = dict((k, mapping[k]) for k in no_expiry_keys)
|
||||
set_arguments = self._get_set_arguments_driver_attr(
|
||||
exclude_expiry=True)
|
||||
self.driver.client.set_multi(new_mapping, **set_arguments)
|
||||
|
||||
if has_expiry_keys:
|
||||
new_mapping = dict((k, mapping[k]) for k in has_expiry_keys)
|
||||
self.driver.set_multi(new_mapping)
|
||||
|
||||
@classmethod
|
||||
def from_config_dict(cls, config_dict, prefix):
|
||||
prefix_len = len(prefix)
|
||||
|
|
@ -120,7 +163,28 @@ class MemcachedBackend(manager.Manager):
|
|||
|
||||
@property
|
||||
def key_mangler(self):
|
||||
return self.driver.key_mangler
|
||||
if self._key_mangler is None:
|
||||
self._key_mangler = self.driver.key_mangler
|
||||
return self._key_mangler
|
||||
|
||||
@key_mangler.setter
|
||||
def key_mangler(self, key_mangler):
|
||||
if callable(key_mangler):
|
||||
self._key_mangler = key_mangler
|
||||
self._rehash_keys()
|
||||
elif key_mangler is None:
|
||||
# NOTE(morganfainberg): Set the hashed key map to the unhashed
|
||||
# list since we no longer have a key_mangler.
|
||||
self._key_mangler = None
|
||||
self.no_expiry_hashed_keys = self.raw_no_expiry_keys
|
||||
else:
|
||||
raise TypeError(_('`key_mangler` functions must be callable.'))
|
||||
|
||||
def _rehash_keys(self):
|
||||
no_expire = set()
|
||||
for key in self.raw_no_expiry_keys:
|
||||
no_expire.add(self._key_mangler(key))
|
||||
self.no_expiry_hashed_keys = no_expire
|
||||
|
||||
def get_mutex(self, key):
|
||||
return MemcachedLock(lambda: self.driver.client, key,
|
||||
|
|
|
|||
|
|
@ -14,10 +14,10 @@
|
|||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
import time
|
||||
import uuid
|
||||
|
||||
from dogpile.cache import api
|
||||
from dogpile.cache.backends import memcached as dogpile_memcached
|
||||
from dogpile.cache import proxy
|
||||
from dogpile.cache import util
|
||||
import mock
|
||||
|
|
@ -87,6 +87,68 @@ class RegionProxy2Fixture(proxy.ProxyBackend):
|
|||
"""A test dogpile.cache proxy that does nothing."""
|
||||
|
||||
|
||||
class TestMemcacheDriver(api.CacheBackend):
|
||||
"""A test dogpile.cache backend that conforms to the mixin-mechanism for
|
||||
overriding set and set_multi methods on dogpile memcached drivers.
|
||||
"""
|
||||
class test_client(object):
|
||||
# FIXME(morganfainberg): Convert this test client over to using mock
|
||||
# and/or mock.MagicMock as appropriate
|
||||
|
||||
def __init__(self):
|
||||
self.__name__ = 'TestingMemcacheDriverClientObject'
|
||||
self.set_arguments_passed = None
|
||||
self.keys_values = {}
|
||||
self.lock_set_time = None
|
||||
self.lock_expiry = None
|
||||
|
||||
def set(self, key, value, **set_arguments):
|
||||
self.keys_values.clear()
|
||||
self.keys_values[key] = value
|
||||
self.set_arguments_passed = set_arguments
|
||||
|
||||
def set_multi(self, mapping, **set_arguments):
|
||||
self.keys_values.clear()
|
||||
self.keys_values = mapping
|
||||
self.set_arguments_passed = set_arguments
|
||||
|
||||
def add(self, key, value, expiry_time):
|
||||
# NOTE(morganfainberg): `add` is used in this case for the
|
||||
# memcache lock testing. If further testing is required around the
|
||||
# actual memcache `add` interface, this method should be
|
||||
# expanded to work more like the actual memcache `add` function
|
||||
if self.lock_expiry is not None and self.lock_set_time is not None:
|
||||
if time.time() - self.lock_set_time < self.lock_expiry:
|
||||
return False
|
||||
self.lock_expiry = expiry_time
|
||||
self.lock_set_time = time.time()
|
||||
return True
|
||||
|
||||
def delete(self, key):
|
||||
# NOTE(morganfainberg): `delete` is used in this case for the
|
||||
# memcache lock testing. If further testing is required around the
|
||||
# actual memcache `delete` interface, this method should be
|
||||
# expanded to work more like the actual memcache `delete` function.
|
||||
self.lock_expiry = None
|
||||
self.lock_set_time = None
|
||||
return True
|
||||
|
||||
def __init__(self, arguments):
|
||||
self.client = self.test_client()
|
||||
self.set_arguments = {}
|
||||
# NOTE(morganfainberg): This is the same logic as the dogpile backend
|
||||
# since we need to mirror that functionality for the `set_argument`
|
||||
# values to appear on the actual backend.
|
||||
if 'memcached_expire_time' in arguments:
|
||||
self.set_arguments['time'] = arguments['memcached_expire_time']
|
||||
|
||||
def set(self, key, value):
|
||||
self.client.set(key, value, **self.set_arguments)
|
||||
|
||||
def set_multi(self, mapping):
|
||||
self.client.set_multi(mapping, **self.set_arguments)
|
||||
|
||||
|
||||
class KVSTest(tests.TestCase):
|
||||
def setUp(self):
|
||||
super(KVSTest, self).setUp()
|
||||
|
|
@ -94,6 +156,10 @@ class KVSTest(tests.TestCase):
|
|||
self.value_foo = uuid.uuid4().hex
|
||||
self.key_bar = 'bar_' + uuid.uuid4().hex
|
||||
self.value_bar = {'complex_data_structure': uuid.uuid4().hex}
|
||||
self.addCleanup(memcached.VALID_DOGPILE_BACKENDS.pop,
|
||||
'TestDriver',
|
||||
None)
|
||||
memcached.VALID_DOGPILE_BACKENDS['TestDriver'] = TestMemcacheDriver
|
||||
|
||||
def _get_kvs_region(self, name=None):
|
||||
if name is None:
|
||||
|
|
@ -140,6 +206,9 @@ class KVSTest(tests.TestCase):
|
|||
kvs.configure('openstack.kvs.Memory')
|
||||
|
||||
self.assertIs(kvs._region.key_mangler, util.sha1_mangle_key)
|
||||
# The backend should also have the keymangler set the same as the
|
||||
# region now.
|
||||
self.assertIs(kvs._region.backend.key_mangler, util.sha1_mangle_key)
|
||||
|
||||
def test_kvs_key_mangler_configuration_backend(self):
|
||||
kvs = self._get_kvs_region()
|
||||
|
|
@ -339,9 +408,9 @@ class KVSTest(tests.TestCase):
|
|||
def test_kvs_memcached_manager_valid_dogpile_memcached_backend(self):
|
||||
kvs = self._get_kvs_region()
|
||||
kvs.configure('openstack.kvs.Memcached',
|
||||
memcached_backend='memcached')
|
||||
memcached_backend='TestDriver')
|
||||
self.assertIsInstance(kvs._region.backend.driver,
|
||||
dogpile_memcached.MemcachedBackend)
|
||||
TestMemcacheDriver)
|
||||
|
||||
def test_kvs_memcached_manager_invalid_dogpile_memcached_backend(self):
|
||||
# Invalid dogpile memcache backend should raise ValueError
|
||||
|
|
@ -351,6 +420,154 @@ class KVSTest(tests.TestCase):
|
|||
backing_store='openstack.kvs.Memcached',
|
||||
memcached_backend=uuid.uuid4().hex)
|
||||
|
||||
def test_kvs_memcache_manager_no_expiry_keys(self):
|
||||
# Make sure the memcache backend recalculates the no-expiry keys
|
||||
# correctly when a key-mangler is set on it.
|
||||
|
||||
def new_mangler(key):
|
||||
return '_mangled_key_' + key
|
||||
|
||||
kvs = self._get_kvs_region()
|
||||
no_expiry_keys = set(['test_key'])
|
||||
kvs.configure('openstack.kvs.Memcached',
|
||||
memcached_backend='TestDriver',
|
||||
no_expiry_keys=no_expiry_keys)
|
||||
calculated_keys = set([kvs._region.key_mangler(key)
|
||||
for key in no_expiry_keys])
|
||||
self.assertIs(kvs._region.backend.key_mangler, util.sha1_mangle_key)
|
||||
self.assertSetEqual(calculated_keys,
|
||||
kvs._region.backend.no_expiry_hashed_keys)
|
||||
self.assertSetEqual(no_expiry_keys,
|
||||
kvs._region.backend.raw_no_expiry_keys)
|
||||
calculated_keys = set([new_mangler(key) for key in no_expiry_keys])
|
||||
kvs._region.backend.key_mangler = new_mangler
|
||||
self.assertSetEqual(calculated_keys,
|
||||
kvs._region.backend.no_expiry_hashed_keys)
|
||||
self.assertSetEqual(no_expiry_keys,
|
||||
kvs._region.backend.raw_no_expiry_keys)
|
||||
|
||||
def test_kvs_memcache_key_mangler_set_to_none(self):
|
||||
kvs = self._get_kvs_region()
|
||||
no_expiry_keys = set(['test_key'])
|
||||
kvs.configure('openstack.kvs.Memcached',
|
||||
memcached_backend='TestDriver',
|
||||
no_expiry_keys=no_expiry_keys)
|
||||
self.assertIs(kvs._region.backend.key_mangler, util.sha1_mangle_key)
|
||||
kvs._region.backend.key_mangler = None
|
||||
self.assertSetEqual(kvs._region.backend.raw_no_expiry_keys,
|
||||
kvs._region.backend.no_expiry_hashed_keys)
|
||||
self.assertIsNone(kvs._region.backend.key_mangler)
|
||||
|
||||
def test_noncallable_key_mangler_set_on_driver_raises_type_error(self):
|
||||
kvs = self._get_kvs_region()
|
||||
kvs.configure('openstack.kvs.Memcached',
|
||||
memcached_backend='TestDriver')
|
||||
self.assertRaises(TypeError,
|
||||
setattr,
|
||||
kvs._region.backend,
|
||||
'key_mangler',
|
||||
'Non-Callable')
|
||||
|
||||
def test_kvs_memcache_set_arguments_and_memcache_expires_ttl(self):
|
||||
# Test the "set_arguments" (arguments passed on all set calls) logic
|
||||
# and the no-expiry-key modifications of set_arguments for the explicit
|
||||
# memcache TTL.
|
||||
self.opt_in_group('kvs', enable_key_mangler=False)
|
||||
kvs = self._get_kvs_region()
|
||||
memcache_expire_time = 86400
|
||||
|
||||
expected_set_args = {'time': memcache_expire_time}
|
||||
expected_no_expiry_args = {}
|
||||
|
||||
expected_foo_keys = [self.key_foo]
|
||||
expected_bar_keys = [self.key_bar]
|
||||
|
||||
mapping_foo = dict([(self.key_foo, self.value_foo)])
|
||||
mapping_bar = dict([(self.key_bar, self.value_bar)])
|
||||
|
||||
kvs.configure(backing_store='openstack.kvs.Memcached',
|
||||
memcached_backend='TestDriver',
|
||||
memcached_expire_time=memcache_expire_time,
|
||||
some_other_arg=uuid.uuid4().hex,
|
||||
no_expiry_keys=[self.key_bar])
|
||||
# Ensure the set_arguments are correct
|
||||
self.assertDictEqual(
|
||||
kvs._region.backend._get_set_arguments_driver_attr(),
|
||||
expected_set_args)
|
||||
|
||||
# Set a key that would have an expiry and verify the correct result
|
||||
# occurred and that the correct set_arguments were passed.
|
||||
kvs.set(self.key_foo, self.value_foo)
|
||||
self.assertDictEqual(
|
||||
kvs._region.backend.driver.client.set_arguments_passed,
|
||||
expected_set_args)
|
||||
self.assertEqual(kvs._region.backend.driver.client.keys_values.keys(),
|
||||
expected_foo_keys)
|
||||
self.assertEqual(
|
||||
kvs._region.backend.driver.client.keys_values[self.key_foo][0],
|
||||
self.value_foo)
|
||||
|
||||
# Set a key that would not have an expiry and verify the correct result
|
||||
# occurred and that the correct set_arguments were passed.
|
||||
kvs.set(self.key_bar, self.value_bar)
|
||||
self.assertDictEqual(
|
||||
kvs._region.backend.driver.client.set_arguments_passed,
|
||||
expected_no_expiry_args)
|
||||
self.assertEqual(kvs._region.backend.driver.client.keys_values.keys(),
|
||||
expected_bar_keys)
|
||||
self.assertEqual(
|
||||
kvs._region.backend.driver.client.keys_values[self.key_bar][0],
|
||||
self.value_bar)
|
||||
|
||||
# set_multi a dict that would have an expiry and verify the correct
|
||||
# result occurred and that the correct set_arguments were passed.
|
||||
kvs.set_multi(mapping_foo)
|
||||
self.assertDictEqual(
|
||||
kvs._region.backend.driver.client.set_arguments_passed,
|
||||
expected_set_args)
|
||||
self.assertEqual(kvs._region.backend.driver.client.keys_values.keys(),
|
||||
expected_foo_keys)
|
||||
self.assertEqual(
|
||||
kvs._region.backend.driver.client.keys_values[self.key_foo][0],
|
||||
self.value_foo)
|
||||
|
||||
# set_multi a dict that would not have an expiry and verify the correct
|
||||
# result occurred and that the correct set_arguments were passed.
|
||||
kvs.set_multi(mapping_bar)
|
||||
self.assertDictEqual(
|
||||
kvs._region.backend.driver.client.set_arguments_passed,
|
||||
expected_no_expiry_args)
|
||||
self.assertEqual(kvs._region.backend.driver.client.keys_values.keys(),
|
||||
expected_bar_keys)
|
||||
self.assertEqual(
|
||||
kvs._region.backend.driver.client.keys_values[self.key_bar][0],
|
||||
self.value_bar)
|
||||
|
||||
def test_memcached_lock_max_lock_attempts(self):
|
||||
kvs = self._get_kvs_region()
|
||||
max_lock_attempts = 1
|
||||
test_key = uuid.uuid4().hex
|
||||
|
||||
kvs.configure(backing_store='openstack.kvs.Memcached',
|
||||
memcached_backend='TestDriver',
|
||||
max_lock_attempts=max_lock_attempts)
|
||||
|
||||
self.assertEqual(kvs._region.backend.max_lock_attempts,
|
||||
max_lock_attempts)
|
||||
# Simple Lock success test
|
||||
with kvs.get_lock(test_key) as lock:
|
||||
kvs.set(test_key, 'testing', lock)
|
||||
|
||||
def lock_within_a_lock(key):
|
||||
with kvs.get_lock(key) as first_lock:
|
||||
kvs.set(test_key, 'lock', first_lock)
|
||||
with kvs.get_lock(key) as second_lock:
|
||||
kvs.set(key, 'lock-within-a-lock', second_lock)
|
||||
|
||||
self.assertRaises(exception.UnexpectedError,
|
||||
lock_within_a_lock,
|
||||
key=test_key)
|
||||
|
||||
|
||||
class TestMemcachedBackend(tests.TestCase):
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue