Merge pull request pandas-dev#772 from shashank88/cache_fixes

Add global settings for caching list_libraries

Author: shashank88

arctic/_cache.py

@@ -3,43 +3,86 @@
 
 from pymongo.errors import OperationFailure
 
-from ._config import CACHE_COLL, CACHE_DB
-
 logger = logging.getLogger(__name__)
 
+CACHE_COLL = 'cache'
+CACHE_DB = 'meta_db'
+CACHE_SETTINGS = 'settings'
+CACHE_SETTINGS_KEY = 'cache'
+"""
+Sample cache_settings collection entry:
+meta_db.cache_settings.insertOne({"type": "cache", "enabled": true, "cache_expiry": 600})
+meta_db.cache_settings.find(): { "_id" : ObjectId("5cd5388b9fddfbe6e968f11b"), "type": "cache", "enabled" : false, "cache_expiry" : 600 }
+"""
+DEFAULT_CACHE_EXPIRY = 3600
+
 
 class Cache:
-    def __init__(self, client, cache_expiry=3600, cache_db=CACHE_DB, cache_col=CACHE_COLL):
+    def __init__(self, client, cache_expiry=DEFAULT_CACHE_EXPIRY, cache_db=CACHE_DB, cache_col=CACHE_COLL):
         self._client = client
         self._cachedb = client[cache_db]
         self._cachecol = None
         try:
             if cache_col not in self._cachedb.list_collection_names():
                 self._cachedb.create_collection(cache_col).create_index("date", expireAfterSeconds=cache_expiry)
         except OperationFailure as op:
-            logging.debug("This is fine if you are not admin. The collection should already be created for you: %s", op)
+            logging.info("This is fine if you are not admin. The collection should already be created for you: %s", op)
 
         self._cachecol = self._cachedb[cache_col]
 
-    def get(self, key, newer_than_secs=-1):
+    def _get_cache_settings(self):
+        try:
+            return self._cachedb[CACHE_SETTINGS].find_one({'type': CACHE_SETTINGS_KEY})
+        except OperationFailure as op:
+            logging.debug("Cannot access %s in db: %s. Error: %s" % (CACHE_SETTINGS, CACHE_DB, op))
+        return None
+
+    def set_caching_state(self, enabled):
+        """
+        Used to enable or disable the caching globally
+        :return:
+        """
+        if not isinstance(enabled, bool):
+            logging.error("Enabled should be a boolean type.")
+            return
+
+        if CACHE_SETTINGS not in self._cachedb.list_collection_names():
+            logging.info("Creating %s collection for cache settings" % CACHE_SETTINGS)
+            self._cachedb[CACHE_SETTINGS].insert_one({
+                'type': CACHE_SETTINGS_KEY,
+                'enabled': enabled,
+                'cache_expiry': DEFAULT_CACHE_EXPIRY
+            })
+        else:
+            self._cachedb[CACHE_SETTINGS].update_one({'type': CACHE_SETTINGS_KEY}, {'$set': {'enabled': enabled}})
+            logging.info("Caching set to: %s" % enabled)
+
+    def _is_not_expired(self, cached_data, newer_than_secs):
+        # Use the expiry period in the settings (or the default) if not overriden by the function argument.
+        if newer_than_secs:
+            expiry_period = newer_than_secs
+        else:
+            cache_settings = self._get_cache_settings()
+            expiry_period = cache_settings['cache_expiry'] if cache_settings else DEFAULT_CACHE_EXPIRY
+
+        return datetime.utcnow() < cached_data['date'] + timedelta(seconds=expiry_period)
+
+    def get(self, key, newer_than_secs=None):
         """
 
         :param key: Key for the dataset. eg. list_libraries.
-        :param newer_than_secs: -1 to indicate use cache if available. Used to indicate what level of staleness
+        :param newer_than_secs: None to indicate use cache if available. Used to indicate what level of staleness
         in seconds is tolerable.
         :return: None unless if there is non stale data present in the cache.
         """
         try:
             if not self._cachecol:
                 # Collection not created or no permissions to read from it.
                 return None
-            coll_data = self._cachecol.find_one({"type": key})
+            cached_data = self._cachecol.find_one({"type": key})
             # Check that there is data in cache and it's not stale.
-            if coll_data and (
-                    newer_than_secs == -1 or
-                    datetime.utcnow() < coll_data['date'] + timedelta(seconds=newer_than_secs)
-            ):
-                return coll_data['data']
+            if cached_data and self._is_not_expired(cached_data, newer_than_secs):
+                return cached_data['data']
         except OperationFailure as op:
             logging.warning("Could not read from cache due to: %s. Ask your admin to give read permissions on %s:%s",
                             op, CACHE_DB, CACHE_COLL)
@@ -83,3 +126,10 @@ def update_item_for_key(self, key, old, new):
         # This op is not atomic, but given the rarity of renaming a lib, it should not cause issues.
         self.delete_item_from_key(key, old)
         self.append(key, new)
+
+    def is_caching_enabled(self):
+        # Caching is enabled unless explicitly disabled.
+        cache_settings = self._get_cache_settings()
+        if cache_settings and not cache_settings['enabled']:
+            return False
+        return True

arctic/_config.py

@@ -94,13 +94,6 @@ class FwPointersCfg(Enum):
 ARCTIC_ASYNC_NWORKERS = os.environ.get('ARCTIC_ASYNC_NWORKERS', 4)
 
 
-# -------------------------------
-# Flag used for indicating caching levels. For now just for list_libraries.
-# -------------------------------
-ENABLE_CACHE = not bool(os.environ.get('DISABLE_CACHE'))
-CACHE_COLL = 'cache'
-CACHE_DB = 'meta_db'
-
 # -------------------------------
 # Flag used to convert byte column/index/column names to unicode when read back.
 # -------------------------------

arctic/arctic.py

@@ -8,7 +8,6 @@
 from six import string_types
 
 from ._cache import Cache
-from ._config import ENABLE_CACHE
 from ._util import indent
 from .auth import authenticate, get_auth
 from .chunkstore import chunkstore
@@ -190,13 +189,20 @@ def __getstate__(self):
     def __setstate__(self, state):
         return Arctic.__init__(self, **state)
 
-    def list_libraries(self, newer_than_secs=-1):
+    def is_caching_enabled(self):
+        """
+        Allows people to enable or disable caching for list_libraries globally.
+        """
+        _ = self._conn  # Ensures the connection exists and cache is initialized with it.
+        return self._cache.is_caching_enabled()
+
+    def list_libraries(self, newer_than_secs=None):
         """
         Returns
         -------
         list of Arctic library names
         """
-        return self._list_libraries_cached(newer_than_secs) if ENABLE_CACHE else self._list_libraries()
+        return self._list_libraries_cached(newer_than_secs) if self.is_caching_enabled() else self._list_libraries()
 
     @mongo_retry
     def _list_libraries(self):
@@ -213,7 +219,7 @@ def _list_libraries(self):
         return libs
 
     # Better to be pessimistic here and not retry.
-    def _list_libraries_cached(self, newer_than_secs=-1):
+    def _list_libraries_cached(self, newer_than_secs=None):
         """
         Returns
         -------
@@ -222,11 +228,14 @@ def _list_libraries_cached(self, newer_than_secs=-1):
         """
         _ = self._conn  # Ensures the connection exists and cache is initialized with it.
         cache_data = self._cache.get('list_libraries', newer_than_secs)
-        if cache_data:
-            logger.debug('Library names are in cache.')
-            return cache_data
-
-        return self._list_libraries()
+        if not cache_data:
+            # Try to refresh the cache.
+            logging.debug("Cache has expired data, fetching from slow path and reloading cache.")
+            libs = self._list_libraries()
+            self._cache.set('list_libraries', libs)
+            return libs
+
+        return cache_data
 
     def reload_cache(self):
         _ = self._conn  # Ensures the connection exists and cache is initialized with it.

tests/integration/test_arctic.py

@@ -315,3 +315,29 @@ def test_deleting_library_removes_it_from_cache(arctic):
     arctic.delete_library('test1')
 
     assert arctic._list_libraries_cached() == arctic._list_libraries() == arctic.list_libraries() == ['test2']
+
+
+def test_disable_cache_by_settings(arctic):
+    lib = 'test1'
+    arctic.initialize_library(lib)
+
+    # Should be enabled by default
+    assert arctic._list_libraries_cached() == arctic._list_libraries()
+
+    arctic._cache.set_caching_state(enabled=False)
+
+    # Should not return cached results now.
+    with patch('arctic.arctic.Arctic._list_libraries', return_value=[lib]) as uncached_list_libraries:
+        with patch('arctic.arctic.Arctic._list_libraries_cached', return_value=[lib]) as cached_list_libraries:
+            arctic.list_libraries()
+            uncached_list_libraries.assert_called()
+            cached_list_libraries.assert_not_called()
+
+    arctic._cache.set_caching_state(enabled=True)
+
+    # Should used cached data again.
+    with patch('arctic.arctic.Arctic._list_libraries', return_value=[lib]) as uncached_list_libraries_e:
+        with patch('arctic.arctic.Arctic._list_libraries_cached', return_value=[lib]) as cached_list_libraries_e:
+            arctic.list_libraries()
+            uncached_list_libraries_e.assert_not_called()
+            cached_list_libraries_e.assert_called()

tests/unit/test_arctic.py

@@ -19,6 +19,7 @@
 def test_arctic_lazy_init():
     with patch('pymongo.MongoClient', return_value=MagicMock(), autospec=True) as mc, \
         patch('arctic.arctic.mongo_retry', side_effect=lambda x: x, autospec=True), \
+        patch('arctic._cache.Cache._is_not_expired', return_value=True), \
         patch('arctic.arctic.get_auth', autospec=True) as ga:
             store = Arctic('cluster')
             assert not mc.called
@@ -30,6 +31,7 @@ def test_arctic_lazy_init():
 def test_arctic_lazy_init_ssl_true():
     with patch('pymongo.MongoClient', return_value=MagicMock(), autospec=True) as mc, \
             patch('arctic.arctic.mongo_retry', side_effect=lambda x: x, autospec=True), \
+            patch('arctic._cache.Cache._is_not_expired', return_value=True), \
             patch('arctic.arctic.get_auth', autospec=True) as ga:
         store = Arctic('cluster', ssl=True)
         assert not mc.called
@@ -48,6 +50,7 @@ def test_arctic_lazy_init_ssl_true():
 def test_connection_passed_warning_raised():
     with patch('pymongo.MongoClient', return_value=MagicMock(), autospec=True), \
          patch('arctic.arctic.mongo_retry', side_effect=lambda x: x, autospec=True), \
+         patch('arctic._cache.Cache._is_not_expired', return_value=True), \
          patch('arctic.arctic.get_auth', autospec=True), \
          patch('arctic.arctic.logger') as lg:
         magic_mock = MagicMock(nodes={("host", "port")})
@@ -62,7 +65,8 @@ def test_connection_passed_warning_raised():
 def test_arctic_auth():
     with patch('pymongo.MongoClient', return_value=MagicMock(), autospec=True), \
         patch('arctic.arctic.mongo_retry', autospec=True), \
-        patch('arctic.arctic.get_auth', autospec=True) as ga:
+         patch('arctic._cache.Cache._is_not_expired', return_value=True), \
+         patch('arctic.arctic.get_auth', autospec=True) as ga:
             ga.return_value = Credential('db', 'admin_user', 'admin_pass')
             store = Arctic('cluster')
             # do something to trigger lazy arctic init
@@ -86,7 +90,8 @@ def test_arctic_auth():
 def test_arctic_auth_custom_app_name():
     with patch('pymongo.MongoClient', return_value=MagicMock(), autospec=True), \
         patch('arctic.arctic.mongo_retry', autospec=True), \
-        patch('arctic.arctic.get_auth', autospec=True) as ga:
+         patch('arctic._cache.Cache._is_not_expired', return_value=True), \
+         patch('arctic.arctic.get_auth', autospec=True) as ga:
             ga.return_value = Credential('db', 'admin_user', 'admin_pass')
             store = Arctic('cluster', app_name=sentinel.app_name)
             # do something to trigger lazy arctic init
@@ -108,7 +113,8 @@ def test_arctic_auth_custom_app_name():
 def test_arctic_connect_hostname():
     with patch('pymongo.MongoClient', return_value=MagicMock(), autospec=True) as mc, \
          patch('arctic.arctic.mongo_retry', autospec=True) as ar, \
-         patch('arctic.arctic.get_mongodb_uri', autospec=True) as gmu:
+            patch('arctic._cache.Cache._is_not_expired', return_value=True), \
+            patch('arctic.arctic.get_mongodb_uri', autospec=True) as gmu:
                 store = Arctic('hostname', socketTimeoutMS=sentinel.socket_timeout,
                                          connectTimeoutMS=sentinel.connect_timeout,
                                          serverSelectionTimeoutMS=sentinel.select_timeout)
@@ -124,6 +130,7 @@ def test_arctic_connect_with_environment_name():
     with patch('pymongo.MongoClient', return_value=MagicMock(), autospec=True) as mc, \
          patch('arctic.arctic.mongo_retry', autospec=True) as ar, \
          patch('arctic.arctic.get_auth', autospec=True), \
+         patch('arctic._cache.Cache._is_not_expired', return_value=True), \
          patch('arctic.arctic.get_mongodb_uri') as gmfe:
             store = Arctic('live', socketTimeoutMS=sentinel.socket_timeout,
                                  connectTimeoutMS=sentinel.connect_timeout,
@@ -453,7 +460,8 @@ def flaky_auth(*args, **kwargs):
 
 def test_reset():
     c = MagicMock()
-    with patch('pymongo.MongoClient', return_value=c, autospec=True) as mc:
+    with patch('pymongo.MongoClient', return_value=c, autospec=True) as mc, \
+            patch('arctic._cache.Cache._is_not_expired', return_value=True):
                 store = Arctic('hostname')
                 # do something to trigger lazy arctic init
                 store.list_libraries()