Improve content provider caching behavior

docker-compose.yml

@@ -53,11 +53,14 @@ services:
       BEHIND_PROXY: "${BEHIND_PROXY}"
       DEFAULT_DATABASE: "${DEFAULT_DATABASE}"
       FLASK_ENV: "${FLASK_ENV}"
+      LOGLEVEL: "${LOGLEVEL}"
+      LOGFILE: "/var/log/api/api.log"
     command: ["python", "app.py"]
     volumes:
       - ./api_cache:/var/api_cache
       - ./server/workers/api/src:/api
       - ./server/workers/common:/api/common
+      - /var/log/api:/var/log/api
     depends_on:
       - redis
       - base

server/workers/api/src/apis/base.py

@@ -9,11 +9,11 @@
 from .request_validators import SearchParamSchema
 from common.utils import get_key, redis_store, get_or_create_contentprovider_lookup
 
-contentprovider_lookup = get_or_create_contentprovider_lookup()
-
 base_ns = Namespace("base", description="BASE API operations")
 search_param_schema = SearchParamSchema()
 
+contentprovider_lookup = get_or_create_contentprovider_lookup(logger=base_ns.logger)
+
 
 base_querymodel = base_ns.model("SearchQuery",
                                {"q": fields.String(example='feminicide',
@@ -84,7 +84,7 @@ def post(self):
         base_ns.logger.debug(d)
         redis_store.rpush("base", json.dumps(d))
         q_len = redis_store.llen("base")
-        base_ns.logger.debug("Queue length: %s %d %s" %("base", q_len, request_id))
+        base_ns.logger.info("Queue length: %s %d %s" %("base", q_len, request_id))
         result = get_key(redis_store, request_id, 300)
         try:
             result, headers = set_response_headers(request.headers["Accept"], params.get("raw"), result, request_id)

server/workers/api/src/apis/create_vis.py

@@ -60,7 +60,7 @@ def post(self):
              "input_data": input_data}
         redis_store.rpush("input_data", json.dumps(d).encode('utf8'))
         q_len = redis_store.llen("input_data")
-        vis_ns.logger.debug("Queue length: %s %d %s" %("input_data", q_len, request_id))
+        vis_ns.logger.info("Queue length: %s %d %s" %("input_data", q_len, request_id))
         result = get_key(redis_store, request_id)
         try:
             result, headers = set_response_headers(request.headers["Accept"], params.get("raw"), result, request_id)

server/workers/api/src/apis/openaire.py

@@ -70,7 +70,7 @@ def post(self):
         openaire_ns.logger.debug(d)
         redis_store.rpush("openaire", json.dumps(d))
         q_len = redis_store.llen("openaire")
-        openaire_ns.logger.debug("Queue length: %s %d %s" % ("openaire", q_len, k))
+        openaire_ns.logger.info("Queue length: %s %d %s" % ("openaire", q_len, k))
         result = get_key(redis_store, k, 300)
         try:
             # Call the utility function once.

server/workers/api/src/apis/orcid.py

@@ -60,7 +60,7 @@ def post(self):
             redis_store.rpush("orcid", json.dumps(task_data))
 
             queue_length = redis_store.llen("orcid")
-            orcid_ns.logger.debug(f"Queue length: orcid {queue_length} {request_id}")
+            orcid_ns.logger.info(f"Queue length: orcid {queue_length} {request_id}")
 
             result = get_key(redis_store, request_id, REDIS_TIMEOUT)
             headers = self.get_response_headers()

server/workers/api/src/apis/pubmed.py

@@ -79,7 +79,7 @@ def post(self):
         pubmed_ns.logger.debug(request_data)
         redis_store.rpush("pubmed", json.dumps(request_data))
         queue_length = redis_store.llen("pubmed")
-        pubmed_ns.logger.debug("Queue length: %s %d %s" % ("pubmed", queue_length, request_id))
+        pubmed_ns.logger.info("Queue length: %s %d %s" % ("pubmed", queue_length, request_id))
 
         # Get the result
         result = get_key(redis_store, request_id)

server/workers/api/src/app.py

@@ -33,10 +33,20 @@ def api_patches(app):
 app = Flask('v1', instance_relative_config=True)
 # Configure logging
 app.logger.setLevel(os.getenv("LOGLEVEL") or logging.DEBUG)
+formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
 handler = logging.StreamHandler(sys.stdout)
-handler.setFormatter(logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s'))
+handler.setFormatter(formatter)
 app.logger.addHandler(handler)
 
+# Optionally also write to a logfile so mtail can turn log lines (e.g.
+# "Queue length: ...") into Prometheus metrics. flask-restx copies these
+# handlers onto each namespace logger at add_namespace() time below.
+logfile = os.getenv("LOGFILE")
+if logfile:
+    file_handler = logging.FileHandler(logfile)
+    file_handler.setFormatter(formatter)
+    app.logger.addHandler(file_handler)
+
 app.wsgi_app = ProxyFix(app.wsgi_app, x_proto=1, x_port=1, x_for=1, x_host=1, x_prefix=1)
 app.wsgi_app = ReverseProxied(app.wsgi_app)
 CORS(app, expose_headers=["Content-Disposition", "Access-Control-Allow-Origin"])
@@ -50,6 +60,12 @@ def api_patches(app):
 api.add_namespace(orcid_ns, path='/orcid')
 api.add_namespace(aquanavi_ns, path='/aquanavi')
 
+# Namespace loggers inherit app.logger's handlers (copied by add_namespace).
+# Disable propagation to the root logger to avoid every line being emitted
+# twice (once via these handlers, once via the Flask default root handler).
+for ns in (base_ns, pubmed_ns, openaire_ns, vis_ns, export_ns, orcid_ns, aquanavi_ns):
+    ns.logger.propagate = False
+
 app.logger.debug(app.config)
 app.logger.debug(app.url_map)
 

server/workers/api/tests/unit/test_contentprovider_cache.py

@@ -0,0 +1,187 @@
+import json
+import time
+import threading
+from contextlib import contextmanager
+
+import pytest
+from redis.exceptions import LockError
+
+import common.utils as cu
+from common.utils import (
+    get_contentprovider_records,
+    get_or_create_contentprovider_lookup,
+    CONTENTPROVIDER_RECORDS_KEY,
+)
+
+RECORDS = [
+    {"name": "Université de Lausanne", "internal_name": "ftunivlausanne"},
+    {"name": "Some Repo", "internal_name": "ftsomerepo"},
+]
+
+
+# --- Minimal in-memory Redis double supporting get / set(ex) / lock -----------
+
+class FakeRedis:
+    def __init__(self, fail_lock=False):
+        self.store = {}
+        self.last_ex = None
+        self._fail_lock = fail_lock
+        self._lock = threading.Lock()
+
+    def get(self, key):
+        return self.store.get(key)
+
+    def set(self, key, value, ex=None):
+        self.store[key] = value
+        self.last_ex = ex
+
+    @contextmanager
+    def lock(self, name, timeout=None, blocking_timeout=None):
+        if self._fail_lock:
+            raise LockError("could not acquire lock")
+        acquired = self._lock.acquire(
+            timeout=blocking_timeout if blocking_timeout is not None else -1
+        )
+        if not acquired:
+            raise LockError("lock acquisition timed out")
+        try:
+            yield
+        finally:
+            self._lock.release()
+
+
+# --- get_contentprovider_records ---------------------------------------------
+
+def test_cache_hit_skips_producer():
+    r = FakeRedis()
+    r.set(CONTENTPROVIDER_RECORDS_KEY, json.dumps(RECORDS))
+    calls = []
+
+    out = get_contentprovider_records(r, lambda: calls.append(1) or [])
+
+    assert out == RECORDS
+    assert calls == []  # producer never called on a warm cache
+
+
+def test_cache_miss_produces_and_caches_with_ttl():
+    r = FakeRedis()
+    calls = []
+
+    def produce():
+        calls.append(1)
+        return RECORDS
+
+    out = get_contentprovider_records(r, produce, ttl=1234)
+
+    assert out == RECORDS
+    assert calls == [1]
+    assert json.loads(r.store[CONTENTPROVIDER_RECORDS_KEY]) == RECORDS
+    assert r.last_ex == 1234
+
+
+def test_single_producer_under_contention():
+    r = FakeRedis()
+    counter = {"n": 0}
+    counter_lock = threading.Lock()
+
+    def produce():
+        with counter_lock:
+            counter["n"] += 1
+        time.sleep(0.2)  # hold the lock long enough for others to contend
+        return RECORDS
+
+    results = []
+
+    def worker():
+        results.append(get_contentprovider_records(r, produce))
+
+    threads = [threading.Thread(target=worker) for _ in range(5)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+
+    assert counter["n"] == 1  # exactly one fetch despite 5 concurrent callers
+    assert all(res == RECORDS for res in results)
+
+
+def test_lock_contention_waits_for_published_cache():
+    r = FakeRedis(fail_lock=True)  # this caller can never acquire the lock
+
+    def publish_later():
+        time.sleep(0.1)
+        r.set(CONTENTPROVIDER_RECORDS_KEY, json.dumps(RECORDS))
+
+    t = threading.Thread(target=publish_later)
+    t.start()
+
+    def produce_should_not_run():
+        raise AssertionError("producer must not run when the lock is held elsewhere")
+
+    out = get_contentprovider_records(r, produce_should_not_run, poll_timeout=5)
+    t.join()
+
+    assert out == RECORDS
+
+
+def test_producer_error_falls_back_to_bundled():
+    r = FakeRedis()
+
+    def produce():
+        raise RuntimeError("boom")
+
+    out = get_contentprovider_records(r, produce)
+
+    assert isinstance(out, list) and len(out) > 0
+    assert {"name", "internal_name"} <= set(out[0].keys())
+
+
+def test_corrupt_cache_value_is_reproduced():
+    r = FakeRedis()
+    r.set(CONTENTPROVIDER_RECORDS_KEY, "not-json{")
+
+    out = get_contentprovider_records(r, lambda: RECORDS)
+
+    assert out == RECORDS
+
+
+def test_cache_hit_logs_debug_trace(caplog):
+    r = FakeRedis()
+    r.set(CONTENTPROVIDER_RECORDS_KEY, json.dumps(RECORDS))
+
+    with caplog.at_level("DEBUG"):
+        get_contentprovider_records(r, lambda: [])
+
+    assert any(
+        "contentprovider-cache: served" in rec.message for rec in caplog.records
+    )
+
+
+def test_fallback_logs_warning(caplog):
+    r = FakeRedis()
+
+    def produce():
+        raise RuntimeError("boom")
+
+    with caplog.at_level("WARNING"):
+        get_contentprovider_records(r, produce)
+
+    assert any(
+        rec.levelname == "WARNING" and "bundled fallback" in rec.message
+        for rec in caplog.records
+    )
+
+
+# --- get_or_create_contentprovider_lookup (API forward map) ------------------
+
+def test_get_or_create_builds_forward_lookup(monkeypatch):
+    monkeypatch.setattr(
+        cu, "get_contentprovider_records", lambda store, fn, **kw: RECORDS
+    )
+
+    lookup = get_or_create_contentprovider_lookup()
+
+    assert lookup == {
+        "ftunivlausanne": "Université de Lausanne",
+        "ftsomerepo": "Some Repo",
+    }

server/workers/base/src/base.py

@@ -27,6 +27,7 @@
 import dateparser
 import sys
 from common.rate_limiter import RateLimiter
+from common.utils import get_contentprovider_records
 
 
 class BaseClient(RWrapper):
@@ -35,8 +36,11 @@ def __init__(self, *args):
         self.rate_limiter = RateLimiter(self.redis_store, "base-ratelimit", 1.5)
 
         try:
-            result = self.get_contentproviders()
-            df = pd.DataFrame(json.loads(result["contentproviders"]))
+            records = get_contentprovider_records(
+                self.redis_store, self._fetch_contentprovider_records,
+                logger=self.logger
+            )
+            df = pd.DataFrame(records)
             df.set_index("name", inplace=True)
             cp_dict = df.internal_name.to_dict()
             self.content_providers = cp_dict
@@ -156,6 +160,13 @@ def enrich_metadata(self, metadata):
         metadata = pd.concat([metadata, enrichment], axis=1)
         return metadata
 
+    def _fetch_contentprovider_records(self):
+        """Run the R fetch and return the parsed list of content provider records."""
+        result = self.get_contentproviders()
+        if result.get("status") == "error":
+            raise RuntimeError("contentproviders fetch returned an error")
+        return json.loads(result["contentproviders"])
+
     def get_contentproviders(self):
         runner = os.path.abspath(os.path.join(self.wd, "run_base_contentproviders.R"))
         cmd = [self.command, runner, self.wd]
@@ -209,7 +220,7 @@ def handle_search(self, request_id, params):
                     "input_data", json.dumps(res).encode("utf8")
                 )
                 q_len = self.redis_store.llen("input_data")
-                self.logger.debug(
+                self.logger.info(
                     "Queue length: %s %d %s" % ("input_data", q_len, request_id)
                 )
         except Exception as e:

server/workers/base/tests/unit/test_base.py

@@ -178,6 +178,18 @@ def dummy_popen_cp(cmd, stdin, stdout, stderr, encoding):
     assert cp_list[0]["name"] == "cp1"
     assert cp_list[0]["internal_name"] == "Provider1"
 
+def test_fetch_contentprovider_records_parses(client_base):
+    # get_contentproviders is stubbed by the fixture to a fixed payload.
+    records = client_base._fetch_contentprovider_records()
+    assert records == [{"name": "cp1", "internal_name": "Provider1"}]
+
+
+def test_fetch_contentprovider_records_raises_on_error(client_base):
+    client_base.get_contentproviders = lambda: {"status": "error"}
+    with pytest.raises(RuntimeError):
+        client_base._fetch_contentprovider_records()
+
+
 # --- Tests for parser functions ---
 
 def test_filter_duplicates():

server/workers/common/common/contentproviders.json

@@ -15067,10 +15067,6 @@
     "name": "Sekigen Akademi Dergileri (SA)",
     "internal_name": "ftsekizgenacad"
   },
-  {
-    "name": "Sekolah Tinggi Agama Islam (STAI) Binamadani Tangerang OJS",
-    "internal_name": "ftstaibinamadani"
-  },
   {
     "name": "Markas Jurnal Sekolah Tinggi Agama Islam Publisistik Thawalib Jakarta",
     "internal_name": "ftstaipublisisti"
@@ -41631,10 +41627,6 @@
     "name": "Универзитет у Нишу",
     "internal_name": "ftunivnisojs"
   },
-  {
-    "name": "MENAdoc (Virtuelle Fachbibliothek Vorderer Orient/Universitäts- und Landesbibliothek Sachsen-Anhalt, Halle)",
-    "internal_name": "ftulbhallemenado"
-  },
   {
     "name": "Hamline University: DigitalCommons@Hamline",
     "internal_name": "fthamlineuniv"