From a3474704c2b6fc53089824bae87d016b00404549 Mon Sep 17 00:00:00 2001 From: David Goetz Date: Wed, 23 Feb 2011 11:44:36 -0800 Subject: [PATCH] ratelimiting does not handle memcache restart --- swift/common/memcached.py | 7 +++ swift/common/middleware/ratelimit.py | 44 ++++++++++--------- test/unit/common/middleware/test_ratelimit.py | 19 ++++++++ test/unit/common/test_memcached.py | 10 +++++ 4 files changed, 60 insertions(+), 20 deletions(-) diff --git a/swift/common/memcached.py b/swift/common/memcached.py index 17378e1aae..8a7ae1a8a5 100644 --- a/swift/common/memcached.py +++ b/swift/common/memcached.py @@ -45,6 +45,10 @@ def md5hash(key): return md5(key).hexdigest() +class MemcacheConnectionError(Exception): + pass + + class MemcacheRing(object): """ Simple, consistent-hashed memcache client. @@ -180,6 +184,7 @@ class MemcacheRing(object): :param delta: amount to add to the value of key (or set as the value if the key is not found) will be cast to an int :param timeout: ttl in memcache + :raises MemcacheConnectionError: """ key = md5hash(key) command = 'incr' @@ -209,6 +214,7 @@ class MemcacheRing(object): return ret except Exception, e: self._exception_occurred(server, e) + raise MemcacheConnectionError("No Memcached connections succeeded.") def decr(self, key, delta=1, timeout=0): """ @@ -220,6 +226,7 @@ class MemcacheRing(object): value to 0 if the key is not found) will be cast to an int :param timeout: ttl in memcache + :raises MemcacheConnectionError: """ self.incr(key, delta=-delta, timeout=timeout) diff --git a/swift/common/middleware/ratelimit.py b/swift/common/middleware/ratelimit.py index 485b1db26e..836cb51bb2 100644 --- a/swift/common/middleware/ratelimit.py +++ b/swift/common/middleware/ratelimit.py @@ -18,6 +18,7 @@ from webob.exc import HTTPNotFound from swift.common.utils import split_path, cache_from_env, get_logger from swift.proxy.server import get_container_memcache_key +from swift.common.memcached import MemcacheConnectionError class MaxSleepTimeHitError(Exception): @@ -136,28 +137,31 @@ class RateLimitMiddleware(object): :param max_rate: maximum rate allowed in requests per second :raises: MaxSleepTimeHitError if max sleep time is exceeded. ''' - now_m = int(round(time.time() * self.clock_accuracy)) - time_per_request_m = int(round(self.clock_accuracy / max_rate)) - running_time_m = self.memcache_client.incr(key, - delta=time_per_request_m) - need_to_sleep_m = 0 - if (now_m - running_time_m > - self.rate_buffer_seconds * self.clock_accuracy): - next_avail_time = int(now_m + time_per_request_m) - self.memcache_client.set(key, str(next_avail_time), - serialize=False) - else: - need_to_sleep_m = \ - max(running_time_m - now_m - time_per_request_m, 0) + try: + now_m = int(round(time.time() * self.clock_accuracy)) + time_per_request_m = int(round(self.clock_accuracy / max_rate)) + running_time_m = self.memcache_client.incr(key, + delta=time_per_request_m) + need_to_sleep_m = 0 + if (now_m - running_time_m > + self.rate_buffer_seconds * self.clock_accuracy): + next_avail_time = int(now_m + time_per_request_m) + self.memcache_client.set(key, str(next_avail_time), + serialize=False) + else: + need_to_sleep_m = \ + max(running_time_m - now_m - time_per_request_m, 0) - max_sleep_m = self.max_sleep_time_seconds * self.clock_accuracy - if max_sleep_m - need_to_sleep_m <= self.clock_accuracy * 0.01: - # treat as no-op decrement time - self.memcache_client.decr(key, delta=time_per_request_m) - raise MaxSleepTimeHitError("Max Sleep Time Exceeded: %s" % - need_to_sleep_m) + max_sleep_m = self.max_sleep_time_seconds * self.clock_accuracy + if max_sleep_m - need_to_sleep_m <= self.clock_accuracy * 0.01: + # treat as no-op decrement time + self.memcache_client.decr(key, delta=time_per_request_m) + raise MaxSleepTimeHitError("Max Sleep Time Exceeded: %s" % + need_to_sleep_m) - return float(need_to_sleep_m) / self.clock_accuracy + return float(need_to_sleep_m) / self.clock_accuracy + except MemcacheConnectionError: + return 0 def handle_ratelimit(self, req, account_name, container_name, obj_name): ''' diff --git a/test/unit/common/middleware/test_ratelimit.py b/test/unit/common/middleware/test_ratelimit.py index ef1abca91e..4afefb0351 100644 --- a/test/unit/common/middleware/test_ratelimit.py +++ b/test/unit/common/middleware/test_ratelimit.py @@ -21,12 +21,14 @@ from webob import Request from swift.common.middleware import ratelimit from swift.proxy.server import get_container_memcache_key +from swift.common.memcached import MemcacheConnectionError class FakeMemcache(object): def __init__(self): self.store = {} + self.error_on_incr = False def get(self, key): return self.store.get(key) @@ -36,6 +38,8 @@ class FakeMemcache(object): return True def incr(self, key, delta=1, timeout=0): + if self.error_on_incr: + raise MemcacheConnectionError('Memcache restarting') self.store[key] = int(self.store.setdefault(key, 0)) + int(delta) if self.store[key] < 0: self.store[key] = 0 @@ -403,6 +407,21 @@ class TestRateLimit(unittest.TestCase): start_response) self._run(make_app_call, num_calls, current_rate) + def test_restarting_memcache(self): + current_rate = 2 + num_calls = 5 + conf_dict = {'account_ratelimit': current_rate} + self.test_ratelimit = ratelimit.filter_factory(conf_dict)(FakeApp()) + ratelimit.http_connect = mock_http_connect(204) + req = Request.blank('/v/a') + req.environ['swift.cache'] = FakeMemcache() + req.environ['swift.cache'].error_on_incr = True + make_app_call = lambda: self.test_ratelimit(req.environ, + start_response) + begin = time.time() + self._run(make_app_call, num_calls, current_rate, check_time=False) + time_took = time.time() - begin + self.assert_(round(time_took, 1) == 0) # no memcache, no limiting if __name__ == '__main__': unittest.main() diff --git a/test/unit/common/test_memcached.py b/test/unit/common/test_memcached.py index 43f11650cf..b54f915dca 100644 --- a/test/unit/common/test_memcached.py +++ b/test/unit/common/test_memcached.py @@ -50,6 +50,7 @@ class MockMemcached(object): self.cache = {} self.down = False self.exc_on_delete = False + self.read_return_none = False def sendall(self, string): if self.down: @@ -110,6 +111,8 @@ class MockMemcached(object): else: self.outbuf += 'NOT_FOUND\r\n' def readline(self): + if self.read_return_none: + return None if self.down: raise Exception('mock is down') if '\n' in self.outbuf: @@ -166,6 +169,9 @@ class TestMemcached(unittest.TestCase): self.assertEquals(memcache_client.get('some_key'), '6') memcache_client.incr('some_key', delta=-15) self.assertEquals(memcache_client.get('some_key'), '0') + mock.read_return_none = True + self.assertRaises(memcached.MemcacheConnectionError, + memcache_client.incr, 'some_key', delta=-15) def test_decr(self): memcache_client = memcached.MemcacheRing(['1.2.3.4:11211']) @@ -179,6 +185,10 @@ class TestMemcached(unittest.TestCase): self.assertEquals(memcache_client.get('some_key'), '11') memcache_client.decr('some_key', delta=15) self.assertEquals(memcache_client.get('some_key'), '0') + mock.read_return_none = True + self.assertRaises(memcached.MemcacheConnectionError, + memcache_client.decr, 'some_key', delta=15) + def test_retry(self): logging.getLogger().addHandler(NullLoggingHandler())