cache configuration improvments (#429)

* cache configuration improvments

* add variable `CACHE_GEOCODING`  - `True` or `False` enable or disable caching for geocoding
* add variable `CACHE_QUERIES` - `True` or `False` enable or disable caching for queries

* fix md indentation

Author: chicco785

docs/manuals/admin/configuration.md

@@ -19,6 +19,8 @@ To configure QuantumLeap you can use the following environment variables:
 | `REDIS_HOST`       | Redis Host              |
 | `REDIS_PORT`       | Redis Port              |
 | `USE_GEOCODING`    | `True` or `False` enable or disable geocoding |
+| `CACHE_GEOCODING`  | `True` or `False` enable or disable caching for geocoding |
+| `CACHE_QUERIES`    | `True` or `False` enable or disable caching for queries |
 | `DEFAULT_CACHE_TTL`| Time to live of metadata cache, default: 60 (seconds) |                              |
 | `QL_CONFIG`        | Pathname for tenant  configuration  |
 | `QL_DEFAULT_DB`    | Default backend: `timescale` or `crate`  |

src/cache/factory.py

@@ -25,6 +25,9 @@ def redis_port(self) -> int:
     def default_ttl(self) -> int:
         return self.env.read(IntVar('DEFAULT_CACHE_TTL', 60))
 
+    def cache_queries(self) -> bool:
+        return self.env.read(BoolVar('CACHE_QUERIES', False))
+
 
 def log():
     return logging.getLogger(__name__)
@@ -51,9 +54,8 @@ def get_cache() -> MaybeCache:
         object otherwise.
     """
     env = CacheEnvReader()
-    if is_cache_available():
-        log().info("Cache env variables set, building a cache.")
-
+    if is_cache_available() and env.cache_queries():
+        log().debug("Cache env variables set, building a cache.")
         return QueryCache(env.redis_host(), env.redis_port(),
                           env.default_ttl())
 

src/geocoding/factory.py

@@ -4,7 +4,6 @@
 from .geocache import GeoCodingCache
 from utils.cfgreader import EnvReader, BoolVar, IntVar, StrVar, MaybeString
 
-
 MaybeGeoCache = Union[GeoCodingCache, None]
 
 
@@ -26,6 +25,9 @@ def __init__(self):
     def use_geocoding(self) -> bool:
         return self.env.read(BoolVar('USE_GEOCODING', False))
 
+    def cache_geocoding(self) -> bool:
+        return self.env.read(BoolVar('CACHE_GEOCODING', False))
+
     def redis_host(self) -> MaybeString:
         return self.env.read(StrVar('REDIS_HOST', None))
 
@@ -48,7 +50,20 @@ def is_geo_coding_available() -> bool:
         use geo-coding.
     """
     env = GeoCodingEnvReader()
-    if env.use_geocoding() and env.redis_host():
+    if env.use_geocoding():
+        return True
+    return False
+
+
+def is_geo_cache_available() -> bool:
+    """
+    Can we use cache? Yes if the Redis host env var is set. No otherwise.
+
+    :return: True or False depending on whether or not we're supposed to
+        use geo-coding.
+    """
+    env = GeoCodingEnvReader()
+    if env.redis_host():
         return True
     return False
 
@@ -62,9 +77,10 @@ def get_geo_cache() -> MaybeGeoCache:
     """
     env = GeoCodingEnvReader()
     if is_geo_coding_available():
-        log().debug("Geo Cache env variables set, building a cache.")
-
-        return GeoCodingCache(env.redis_host(), env.redis_port())
+        log().debug("Geo Cache env variables set, try to build a cache.")
+        if env.cache_geocoding() and is_geo_cache_available():
+            return GeoCodingCache(env.redis_host(), env.redis_port())
+        log().warning("Geo Cache is not enabled, check env variables.")
 
-    log().debug("Geo Cache env variables indicate cache should not be used.")
+    log().debug("Geo Cache is not enabled")
     return None

src/reporter/tests/docker-compose.yml

@@ -40,6 +40,7 @@ services:
     environment:
       - USE_GEOCODING=False
       - REDIS_HOST=redis
+      - CACHE_QUERIES=True
       - LOGLEVEL=INFO
       - POSTGRES_HOST=timescale
       - QL_CONFIG=/config/ql-config.yml

src/tests/benchmark/docker-compose.yml

@@ -45,6 +45,7 @@ services:
     environment:
       - CRATE_HOST=${CRATE_HOST:-crate}
       - USE_GEOCODING=False
+      - CACHE_QUERIES=True
       - REDIS_HOST=redis
       - REDIS_PORT=6379
       - LOGLEVEL=INFO

src/tests/docker-compose.yml

@@ -33,6 +33,7 @@ services:
     environment:
       - CRATE_HOST=${CRATE_HOST:-crate}
       - USE_GEOCODING=False
+      - CACHE_QUERIES=True
       - REDIS_HOST=redis
       - REDIS_PORT=6379
       - LOGLEVEL=INFO

src/translators/sql_translator.py

@@ -131,8 +131,8 @@ def __init__(self, host, port, db_name):
 
     def dispose(self):
         dt = datetime.now() - self.start_time
-        time_difference = (
-                                      dt.days * 24 * 60 * 60 + dt.seconds) * 1000 + dt.microseconds / 1000.0
+        time_difference = (dt.days * 24 * 60 * 60 + dt.seconds)\
+            * 1000 + dt.microseconds / 1000.0
         self.logger.debug("Translation completed | time={} msec".format(
             str(time_difference)))
 
@@ -143,8 +143,8 @@ def _refresh(self, entity_types, fiware_service=None):
         """
         Used for testing purposes only!
         Refreshing ensures a query after an insert retrieves the inserted data.
-        :param entity_types: list(str) list of entity types whose tables will be
-         refreshed
+        :param entity_types: list(str) list of entity types whose tables will
+         be refreshed
         """
         table_names = [self._et2tn(et, fiware_service) for et in entity_types]
         table_names.append(METADATA_TABLE_NAME)
@@ -365,8 +365,8 @@ def _insert_entity_rows(self, table_name: str, col_names: List[str],
             start_time = datetime.now()
             self.cursor.executemany(stmt, rows)
             dt = datetime.now() - start_time
-            time_difference = (
-                                          dt.days * 24 * 60 * 60 + dt.seconds) * 1000 + dt.microseconds / 1000.0
+            time_difference = (dt.days * 24 * 60 * 60 + dt.seconds)\
+                * 1000 + dt.microseconds / 1000.0
             self.logger.debug("Query completed | time={} msec".format(
                 str(time_difference)))
         except Exception as e:
@@ -426,8 +426,8 @@ def _should_insert_original_entities(self,
     def _insert_original_entities_in_failed_batch(
             self, table_name: str, entities: List[dict],
             insert_error: Exception):
-        cols = f"{ENTITY_ID_COL}, {ENTITY_TYPE_COL}, {self.TIME_INDEX_NAME}" + \
-               f", {ORIGINAL_ENTITY_COL}"
+        cols = f"{ENTITY_ID_COL}, {ENTITY_TYPE_COL}, {self.TIME_INDEX_NAME}" \
+               + f", {ORIGINAL_ENTITY_COL}"
         stmt = f"insert into {table_name} ({cols}) values (?, ?, ?, ?)"
         tix = current_timex()
         batch_id = uuid4().hex
@@ -438,7 +438,8 @@ def _insert_original_entities_in_failed_batch(
         self.cursor.executemany(stmt, rows)
 
     def _attr_is_structured(self, a):
-        if 'value' in a and a['value'] is not None and isinstance(a['value'], dict):
+        if 'value' in a and a['value'] is not None \
+                and isinstance(a['value'], dict):
             self.logger.debug("attribute {} has 'value' attribute of type dict"
                               .format(a))
             return True
@@ -448,8 +449,8 @@ def _attr_is_structured(self, a):
     @staticmethod
     def is_text(attr_type):
         # TODO: verify: same logic in two different places!
-        # The above kinda reproduces the tests done by the translator, we should
-        # factor this logic out and keep it in just one place!
+        # The above kinda reproduces the tests done by the translator,
+        # we should factor this logic out and keep it in just one place!
         return attr_type == NGSI_TEXT or attr_type not in NGSI_TO_SQL
 
     def _preprocess_values(self, e, original_attrs, col_names,
@@ -1352,20 +1353,29 @@ def _execute_query_via_cache(self, tenant_name, key, stmt, parameters=None,
                     res = pickle.loads(value)
                     return res
             except Exception as e:
-                self.logger.warning(str(e), exc_info=True)
+                self.logger.warning("Caching not available, metadata data may "
+                                    "not be consistent: " + str(e),
+                                    exc_info=True)
 
         self.cursor.execute(stmt, parameters)
         res = self.cursor.fetchall()
-        if res:
-            self._cache(tenant_name, key, res, ex)
+        if res and self.cache:
+            try:
+                self._cache(tenant_name, key, res, ex)
+            except Exception as e:
+                self.logger.warning("Caching not available, metadata data may "
+                                    "not be consistent: " + str(e),
+                                    exc_info=True)
         return res
 
     def _is_query_in_cache(self, tenant_name, key):
         if self.cache:
             try:
                 return self.cache.exists(tenant_name, key)
             except Exception as e:
-                self.logger.warning(str(e), exc_info=True)
+                self.logger.warning("Caching not available, metadata data may "
+                                    "not be consistent: " + str(e),
+                                    exc_info=True)
         return False
 
     def _cache(self, tenant_name, key, value=None, ex=None):
@@ -1375,14 +1385,18 @@ def _cache(self, tenant_name, key, value=None, ex=None):
                     value = pickle.dumps(value)
                 self.cache.put(tenant_name, key, value, ex)
             except Exception as e:
-                self.logger.warning(str(e), exc_info=True)
+                self.logger.warning("Caching not available, metadata data may "
+                                    "not be consistent: " + str(e),
+                                    exc_info=True)
 
     def _remove_from_cache(self, tenant_name, key):
         if self.cache:
             try:
                 self.cache.delete(tenant_name, key)
             except Exception as e:
-                self.logger.warning(str(e), exc_info=True)
+                self.logger.warning("Caching not available, metadata data may "
+                                    "not be consistent: " + str(e),
+                                    exc_info=True)
 
 
 class QueryCacheManager(Borg):
@@ -1394,7 +1408,8 @@ def __init__(self):
             try:
                 self.cache = get_cache()
             except Exception as e:
-                self.logger.warning(str(e), exc_info=True)
+                self.logger.warning("Caching not available:" + str(e),
+                                    exc_info=True)
 
     def get_query_cache(self):
         return self.cache