Complete rate-limiting for Mapzen & Here gecoding

ServiceManager class has been introduced to handle service configuration at
SQL level (with a LegacyServiceManager alternative for non-refactored services).
These new classes take the responsibility of rate limits and quota checking.

Tests have been added for ServiceManager and rate limits, but currently they
check only the limits configuration since Lua support would be needed
to use rratelimit with MockRedis.

server/lib/python/cartodb_services/cartodb_services/metrics/config.py

@@ -523,15 +523,15 @@ class ServicesDBConfig:
                 self._data_observatory_connection_str = do_conf['connection']['production']
 
     def _get_rate_limits_config(self):
-        self._rate_limits = self._get_conf('rate_limits', default={})
+        self._rate_limits = json.loads(self._get_conf('rate_limits', default='{}'))
 
-    def _get_conf(self, key, default='raise'):
+    def _get_conf(self, key, default=KeyError):
         try:
             sql = "SELECT cartodb.CDB_Conf_GetConf('{0}') as conf".format(key)
             conf = self._db_conn.execute(sql, 1)
             return conf[0]['conf']
         except Exception as e:
-            if (default != 'raise'):
+            if (default != KeyError):
                 return default
             raise ConfigException("Error trying to get config for {0}: {1}".format(key, e))
 

server/lib/python/cartodb_services/cartodb_services/refactor/config/rate_limits.py

@@ -50,14 +50,15 @@ class RateLimitsConfigBuilder(object):
     def get(self):
         # Order of precedence is user_conf, org_conf, server_conf
 
-        rate_limit_key = "{0}_rate_limit".format(service)
+        rate_limit_key = "{0}_rate_limit".format(self._service)
         rate_limit_json = self._user_conf.get(rate_limit_key, None) or self._org_conf.get(rate_limit_key, None)
         if (rate_limit_json):
             rate_limit = rate_limit_json and json.loads(rate_limit_json)
         else:
-            rate_limit = self._server_conf.get('rate_limits', {}).get(service, {})
+            rate_limit = self._server_conf.get('rate_limits', {}).get(self._service, {})
 
         return RateLimitsConfig(self._service,
                                 self._username,
-                                self._limit,
-                                self._period)
+                                rate_limit.get('limit', None),
+                                rate_limit.get('period', None))
+

server/lib/python/cartodb_services/cartodb_services/refactor/core/interfaces.py

@@ -6,6 +6,6 @@ class ConfigBackendInterface(object):
     __metaclass__ = abc.ABCMeta
 
     @abc.abstractmethod
-    def get(self, key):
+    def get(self, key, default=None):
         """Return a value based on the key supplied from some storage"""
         pass

server/lib/python/cartodb_services/cartodb_services/refactor/storage/mem_config.py

@@ -5,8 +5,8 @@ class InMemoryConfigStorage(ConfigBackendInterface):
     def __init__(self, config_hash={}):
         self._config_hash = config_hash
 
-    def get(self, key):
+    def get(self, key, default=None):
         try:
             return self._config_hash[key]
         except KeyError:
-            return None
+            return default

server/lib/python/cartodb_services/cartodb_services/refactor/storage/null_config.py

@@ -2,5 +2,5 @@ from ..core.interfaces import ConfigBackendInterface
 
 class NullConfigStorage(ConfigBackendInterface):
 
-    def get(self, key):
-        return None
+    def get(self, key, default=None):
+        return default

server/lib/python/cartodb_services/cartodb_services/refactor/storage/redis_config.py

@@ -9,11 +9,13 @@ class RedisConfigStorage(ConfigBackendInterface):
         self._config_key = config_key
         self._data = None
 
-    def get(self, key):
+    def get(self, key, default=KeyError):
         if not self._data:
             self._data = self._connection.hgetall(self._config_key)
-        return self._data[key]
-
+        if (default == KeyError):
+            return self._data[key]
+        else:
+            return self._data.get(key, default)
 
 class RedisUserConfigStorageBuilder(object):
     def __init__(self, redis_connection, username):

server/lib/python/cartodb_services/cartodb_services/refactor/storage/server_config.py

@@ -4,11 +4,18 @@ from ..core.interfaces import ConfigBackendInterface
 
 class InDbServerConfigStorage(ConfigBackendInterface):
 
-    def get(self, key):
+    def get(self, key, default=None):
         sql = "SELECT cdb_dataservices_server.cdb_conf_getconf('{0}') as conf".format(key)
         rows = cartodb_services.plpy.execute(sql, 1)
-        json_output = rows[0]['conf']
-        if json_output:
+        json_output = None
+        try:
+            json_output = rows[0]['conf']
+        except IndexError:
+            pass
+        if (json_output):
             return json.loads(json_output)
         else:
-            return None
+            if (default == KeyError):
+                raise KeyError
+            else:
+                return default

server/lib/python/cartodb_services/cartodb_services/tools/__init__.py

@@ -2,3 +2,6 @@ from redis_tools import RedisConnection, RedisDBConfig
 from coordinates import Coordinate
 from polyline import PolyLine
 from log import Logger, LoggerConfig
+from rate_limiter import RateLimiter
+from exceptions import RateLimitExceeded
+from service_manager import ServiceManager, LegacyServiceManager

server/lib/python/cartodb_services/cartodb_services/tools/exceptions.py

@@ -0,0 +1,3 @@
+class RateLimitExceeded(Exception):
+    def __str__(self):
+            return repr('Rate limit exceeded')

server/lib/python/cartodb_services/cartodb_services/tools/rate_limiter.py

@@ -4,13 +4,14 @@ class RateLimiter:
 
     def __init__(self, rate_limits_config, redis_connection):
         self._config = rate_limits_config
+        self._limiter = None
         if (self._config.is_limited()):
-            self._limiter = Limiter(rdis_connection,
+            self._limiter = Limiter(redis_connection,
                                     action=self._config.service,
                                     limit=self._config.limit,
                                     period=self._config.period)
 
-    def check():
+    def check(self):
         ok = True
         if (self._limiter):
             ok = self._limiter.checked_insert(self._config.username)

server/lib/python/cartodb_services/cartodb_services/tools/service_manager.py

@@ -0,0 +1,70 @@
+from exceptions import RateLimitExceeded
+
+class ServiceManagerBase:
+    def check(self):
+        if not self.rate_limiter.check():
+            raise RateLimitExceeded()
+        if not self.quota_service.check_user_quota():
+            raise Exception('You have reached the limit of your quota')
+
+    @property
+    def config(self):
+        return self.config
+
+    @property
+    def quota_service(self):
+        return self.quota_service
+
+    @property
+    def logger(self):
+        return self.logger
+
+from cartodb_services.metrics import QuotaService
+from cartodb_services.tools import Logger,LoggerConfig
+from cartodb_services.tools import RateLimiter
+from cartodb_services.refactor.config.rate_limits import RateLimitsConfig
+
+class LegacyServiceManager(ServiceManagerBase):
+
+    def __init__(self, service, username, orgname, gd):
+        redis_conn = gd["redis_connection_{0}".format(username)]['redis_metrics_connection']
+        self.config = gd["user_{0}_config_{1}".format(service, username)]
+        logger_config = gd["logger_config"]
+        self.logger = Logger(logger_config)
+
+        rate_limits_config = RateLimitsConfig(service,
+                                                username,
+                                                self.config.rate_limit.get('limit'),
+                                                self.config.rate_limit.get('period'))
+        self.rate_limiter = RateLimiter(rate_limits_config, redis_conn)
+        self.quota_service = QuotaService(self.config, redis_conn)
+
+from cartodb_services.metrics import QuotaService
+from cartodb_services.tools import Logger
+from cartodb_services.tools import RateLimiter
+from cartodb_services.refactor.tools.logger import LoggerConfigBuilder
+from cartodb_services.refactor.core.environment import ServerEnvironmentBuilder
+from cartodb_services.refactor.backend.server_config import ServerConfigBackendFactory
+from cartodb_services.refactor.backend.user_config import UserConfigBackendFactory
+from cartodb_services.refactor.backend.org_config import OrgConfigBackendFactory
+from cartodb_services.refactor.backend.redis_metrics_connection import RedisMetricsConnectionFactory
+from cartodb_services.refactor.config.rate_limits import RateLimitsConfigBuilder
+
+class ServiceManager(ServiceManagerBase):
+
+    def __init__(self, service, config_builder, username, orgname):
+        server_config_backend = ServerConfigBackendFactory().get()
+        environment = ServerEnvironmentBuilder(server_config_backend).get()
+        user_config_backend = UserConfigBackendFactory(username, environment, server_config_backend).get()
+        org_config_backend = OrgConfigBackendFactory(orgname, environment, server_config_backend).get()
+
+        logger_config = LoggerConfigBuilder(environment, server_config_backend).get()
+        self.logger = Logger(logger_config)
+
+        self.config = config_builder(server_config_backend, user_config_backend, org_config_backend, username, orgname).get()
+        rate_limit_config = RateLimitsConfigBuilder(server_config_backend, user_config_backend, org_config_backend, service=service, user=username, org=orgname).get()
+
+        redis_metrics_connection = RedisMetricsConnectionFactory(environment, server_config_backend).get()
+
+        self.rate_limiter = RateLimiter(rate_limit_config, redis_metrics_connection)
+        self.quota_service = QuotaService(self.config, redis_metrics_connection)

server/lib/python/cartodb_services/test/test_servicemanager.py

@@ -0,0 +1,218 @@
+from test_helper import *
+from unittest import TestCase
+from mock import Mock, MagicMock, patch
+from nose.tools import assert_raises, assert_not_equal, assert_equal
+
+from datetime import datetime, date
+from cartodb_services.tools import ServiceManager, LegacyServiceManager
+from mockredis import MockRedis
+import cartodb_services
+from cartodb_services.metrics import GeocoderConfig
+from cartodb_services.refactor.service.mapzen_geocoder_config import MapzenGeocoderConfigBuilder
+from cartodb_services.refactor.backend.redis_metrics_connection import RedisConnectionBuilder
+from cartodb_services.tools import RateLimitExceeded
+
+LUA_AVAILABLE_FOR_MOCKREDIS = False
+
+class MockRedisWithVersionInfo(MockRedis):
+    def info(self):
+        return {'redis_version': '3.0.2'}
+
+class TestServiceManager(TestCase):
+
+    def setUp(self):
+        plpy_mock_config()
+        cartodb_services.init(plpy_mock, _GD={})
+        self.username = 'test_user'
+        self.orgname = 'test_org'
+        self.redis_conn = MockRedisWithVersionInfo()
+        build_redis_user_config(self.redis_conn, self.username, 'geocoding')
+        build_redis_org_config(self.redis_conn, self.orgname, 'geocoding', provider='mapzen')
+        self.environment = 'production'
+        plpy_mock._define_result("CDB_Conf_GetConf\('server_conf'\)", [{'conf': '{"environment": "production"}'}])
+        plpy_mock._define_result("CDB_Conf_GetConf\('redis_metadata_config'\)", [{'conf': '{"redis_host":"localhost","redis_port":"6379"}'}])
+        plpy_mock._define_result("CDB_Conf_GetConf\('redis_metrics_config'\)", [{'conf': '{"redis_host":"localhost","redis_port":"6379"}'}])
+
+    def check_rate_limit(self, service_manager, n, active=True):
+        if LUA_AVAILABLE_FOR_MOCKREDIS:
+            for _ in xrange(n):
+                service_manager.check()
+            if active:
+                with assert_raises(RateLimitExceeded):
+                    service_manager.check()
+            else:
+                service_manager.check()
+        else:
+            # rratelimit doesn't work with MockRedis because it needs Lua support
+            # so, we'll simply perform some sanity check on the configuration of the rate limiter
+            if active:
+                assert_equal(service_manager.rate_limiter._config.is_limited(), True)
+                assert_equal(service_manager.rate_limiter._config.limit, n)
+            else:
+                assert not service_manager.rate_limiter._config.is_limited()
+
+    def test_legacy_service_manager(self):
+        config = GeocoderConfig(self.redis_conn, plpy_mock, self.username, self.orgname,  'mapzen')
+        config_cache = {
+          'redis_connection_test_user' : { 'redis_metrics_connection': self.redis_conn },
+          'user_geocoder_config_test_user' : config,
+          'logger_config' : Mock(min_log_level='debug', log_file_path=None, rollbar_api_key=None, environment=self.environment)
+        }
+        service_manager = LegacyServiceManager('geocoder', self.username, self.orgname, config_cache)
+        service_manager.check()
+        assert_equal(service_manager.config.service_type, 'geocoder_mapzen')
+
+    def test_service_manager(self):
+        with patch.object(RedisConnectionBuilder,'get') as get_fn:
+            get_fn.return_value = self.redis_conn
+            service_manager = ServiceManager('geocoder', MapzenGeocoderConfigBuilder, self.username, self.orgname)
+            service_manager.check()
+            assert_equal(service_manager.config.service_type, 'geocoder_mapzen')
+
+    def test_no_rate_limit_by_default(self):
+        with patch.object(RedisConnectionBuilder,'get') as get_fn:
+            get_fn.return_value = self.redis_conn
+            service_manager = ServiceManager('geocoder', MapzenGeocoderConfigBuilder, self.username, self.orgname)
+            self.check_rate_limit(service_manager, 3, False)
+
+    def test_no_legacy_rate_limit_by_default(self):
+        config = GeocoderConfig(self.redis_conn, plpy_mock, self.username, self.orgname,  'mapzen')
+        config_cache = {
+          'redis_connection_test_user' : { 'redis_metrics_connection': self.redis_conn },
+          'user_geocoder_config_test_user' : config,
+          'logger_config' : Mock(min_log_level='debug', log_file_path=None, rollbar_api_key=None, environment=self.environment)
+        }
+        service_manager = LegacyServiceManager('geocoder', self.username, self.orgname, config_cache)
+        self.check_rate_limit(service_manager, 3, False)
+
+    def test_legacy_server_rate_limit(self):
+        rate_limits = '{"geocoder":{"limit":"3","period":3600}}'
+        plpy_mock._define_result("CDB_Conf_GetConf\('rate_limits'\)", [{'conf': rate_limits}])
+        config = GeocoderConfig(self.redis_conn, plpy_mock, self.username, self.orgname,  'mapzen')
+        config_cache = {
+          'redis_connection_test_user' : { 'redis_metrics_connection': self.redis_conn },
+          'user_geocoder_config_test_user' : config,
+          'logger_config' : Mock(min_log_level='debug', log_file_path=None, rollbar_api_key=None, environment=self.environment)
+        }
+        service_manager = LegacyServiceManager('geocoder', self.username, self.orgname, config_cache)
+        self.check_rate_limit(service_manager, 3)
+        plpy_mock._define_result("CDB_Conf_GetConf\('rate_limits'\)", [])
+
+    def test_server_rate_limit(self):
+        rate_limits = '{"geocoder":{"limit":"3","period":3600}}'
+        plpy_mock._define_result("CDB_Conf_GetConf\('rate_limits'\)", [{'conf': rate_limits}])
+        with patch.object(RedisConnectionBuilder,'get') as get_fn:
+            get_fn.return_value = self.redis_conn
+            service_manager = ServiceManager('geocoder', MapzenGeocoderConfigBuilder, self.username, self.orgname)
+            self.check_rate_limit(service_manager, 3)
+        plpy_mock._define_result("CDB_Conf_GetConf\('rate_limits'\)", [])
+
+    def test_user_rate_limit(self):
+        user_redis_name = "rails:users:{0}".format(self.username)
+        rate_limits = '{"limit":"3","period":3600}'
+        self.redis_conn.hset(user_redis_name, 'geocoder_rate_limit', rate_limits)
+        with patch.object(RedisConnectionBuilder,'get') as get_fn:
+            get_fn.return_value = self.redis_conn
+            service_manager = ServiceManager('geocoder', MapzenGeocoderConfigBuilder, self.username, self.orgname)
+            self.check_rate_limit(service_manager, 3)
+        self.redis_conn.hdel(user_redis_name, 'geocoder_rate_limit')
+
+    def test_legacy_user_rate_limit(self):
+        user_redis_name = "rails:users:{0}".format(self.username)
+        rate_limits = '{"limit":"3","period":3600}'
+        self.redis_conn.hset(user_redis_name, 'geocoder_rate_limit', rate_limits)
+        config = GeocoderConfig(self.redis_conn, plpy_mock, self.username, self.orgname,  'mapzen')
+        config_cache = {
+          'redis_connection_test_user' : { 'redis_metrics_connection': self.redis_conn },
+          'user_geocoder_config_test_user' : config,
+          'logger_config' : Mock(min_log_level='debug', log_file_path=None, rollbar_api_key=None, environment=self.environment)
+        }
+        service_manager = LegacyServiceManager('geocoder', self.username, self.orgname, config_cache)
+        self.check_rate_limit(service_manager, 3)
+        self.redis_conn.hdel(user_redis_name, 'geocoder_rate_limit')
+
+    def test_org_rate_limit(self):
+        org_redis_name = "rails:orgs:{0}".format(self.orgname)
+        rate_limits = '{"limit":"3","period":3600}'
+        self.redis_conn.hset(org_redis_name, 'geocoder_rate_limit', rate_limits)
+        with patch.object(RedisConnectionBuilder,'get') as get_fn:
+            get_fn.return_value = self.redis_conn
+            service_manager = ServiceManager('geocoder', MapzenGeocoderConfigBuilder, self.username, self.orgname)
+            self.check_rate_limit(service_manager, 3)
+        self.redis_conn.hdel(org_redis_name, 'geocoder_rate_limit')
+
+    def test_legacy_org_rate_limit(self):
+        org_redis_name = "rails:orgs:{0}".format(self.orgname)
+        rate_limits = '{"limit":"3","period":3600}'
+        self.redis_conn.hset(org_redis_name, 'geocoder_rate_limit', rate_limits)
+        config = GeocoderConfig(self.redis_conn, plpy_mock, self.username, self.orgname,  'mapzen')
+        config_cache = {
+          'redis_connection_test_user' : { 'redis_metrics_connection': self.redis_conn },
+          'user_geocoder_config_test_user' : config,
+          'logger_config' : Mock(min_log_level='debug', log_file_path=None, rollbar_api_key=None, environment=self.environment)
+        }
+        service_manager = LegacyServiceManager('geocoder', self.username, self.orgname, config_cache)
+        self.check_rate_limit(service_manager, 3)
+        self.redis_conn.hdel(org_redis_name, 'geocoder_rate_limit')
+
+    def test_user_rate_limit_precedence_over_org(self):
+        org_redis_name = "rails:orgs:{0}".format(self.orgname)
+        org_rate_limits = '{"limit":"1000","period":3600}'
+        self.redis_conn.hset(org_redis_name, 'geocoder_rate_limit', org_rate_limits)
+        user_redis_name = "rails:users:{0}".format(self.username)
+        user_rate_limits = '{"limit":"3","period":3600}'
+        self.redis_conn.hset(user_redis_name, 'geocoder_rate_limit', user_rate_limits)
+        with patch.object(RedisConnectionBuilder,'get') as get_fn:
+            get_fn.return_value = self.redis_conn
+            service_manager = ServiceManager('geocoder', MapzenGeocoderConfigBuilder, self.username, self.orgname)
+            self.check_rate_limit(service_manager, 3)
+        self.redis_conn.hdel(org_redis_name, 'geocoder_rate_limit')
+        self.redis_conn.hdel(user_redis_name, 'geocoder_rate_limit')
+
+    def test_org_rate_limit_precedence_over_server(self):
+        server_rate_limits = '{"geocoder":{"limit":"1000","period":3600}}'
+        plpy_mock._define_result("CDB_Conf_GetConf\('rate_limits'\)", [{'conf': server_rate_limits}])
+        org_redis_name = "rails:orgs:{0}".format(self.orgname)
+        org_rate_limits = '{"limit":"3","period":3600}'
+        self.redis_conn.hset(org_redis_name, 'geocoder_rate_limit', org_rate_limits)
+        with patch.object(RedisConnectionBuilder,'get') as get_fn:
+            get_fn.return_value = self.redis_conn
+            service_manager = ServiceManager('geocoder', MapzenGeocoderConfigBuilder, self.username, self.orgname)
+            self.check_rate_limit(service_manager, 3)
+        self.redis_conn.hdel(org_redis_name, 'geocoder_rate_limit')
+        plpy_mock._define_result("CDB_Conf_GetConf\('rate_limits'\)", [])
+
+    def test_legacy_user_rate_limit_precedence_over_org(self):
+        org_redis_name = "rails:orgs:{0}".format(self.orgname)
+        org_rate_limits = '{"limit":"1000","period":3600}'
+        self.redis_conn.hset(org_redis_name, 'geocoder_rate_limit', org_rate_limits)
+        user_redis_name = "rails:users:{0}".format(self.username)
+        user_rate_limits = '{"limit":"3","period":3600}'
+        self.redis_conn.hset(user_redis_name, 'geocoder_rate_limit', user_rate_limits)
+        config = GeocoderConfig(self.redis_conn, plpy_mock, self.username, self.orgname,  'mapzen')
+        config_cache = {
+          'redis_connection_test_user' : { 'redis_metrics_connection': self.redis_conn },
+          'user_geocoder_config_test_user' : config,
+          'logger_config' : Mock(min_log_level='debug', log_file_path=None, rollbar_api_key=None, environment=self.environment)
+        }
+        service_manager = LegacyServiceManager('geocoder', self.username, self.orgname, config_cache)
+        self.check_rate_limit(service_manager, 3)
+        self.redis_conn.hdel(org_redis_name, 'geocoder_rate_limit')
+        self.redis_conn.hdel(user_redis_name, 'geocoder_rate_limit')
+
+    def test_legacy_org_rate_limit_precedence_over_server(self):
+        server_rate_limits = '{"geocoder":{"limit":"1000","period":3600}}'
+        plpy_mock._define_result("CDB_Conf_GetConf\('rate_limits'\)", [{'conf': server_rate_limits}])
+        org_redis_name = "rails:orgs:{0}".format(self.orgname)
+        org_rate_limits = '{"limit":"3","period":3600}'
+        self.redis_conn.hset(org_redis_name, 'geocoder_rate_limit', org_rate_limits)
+        config = GeocoderConfig(self.redis_conn, plpy_mock, self.username, self.orgname,  'mapzen')
+        config_cache = {
+          'redis_connection_test_user' : { 'redis_metrics_connection': self.redis_conn },
+          'user_geocoder_config_test_user' : config,
+          'logger_config' : Mock(min_log_level='debug', log_file_path=None, rollbar_api_key=None, environment=self.environment)
+        }
+        service_manager = LegacyServiceManager('geocoder', self.username, self.orgname, config_cache)
+        self.check_rate_limit(service_manager, 3)
+        self.redis_conn.hdel(org_redis_name, 'geocoder_rate_limit')
+        plpy_mock._define_result("CDB_Conf_GetConf\('rate_limits'\)", [])