Add a basic caching system for some datanommer values

Signed-off-by: Aurélien Bompard <[email protected]>

devel/ansible/roles/dev/files/config.toml

@@ -48,6 +48,13 @@ id_provider_hostname = "id.tinystage.test"
 fasjson_base_url = "https://fasjson.tinystage.test/fasjson"
 
 
+# Cache configuation
+[consumer_config.cache]
+backend = "dogpile.cache.dbm"
+expiration_time = 10000
+[consumer_config.cache.arguments]
+filename = /home/vagrant/cache-fedbadges.dbm
+
 # This is a set of data that tells our consumer what Open Badges Issuer
 # should be kept as the issuer of all the badges we create.
 [consumer_config.badge_issuer]

fedbadges.toml.example

@@ -47,6 +47,10 @@ distgit_hostname = "src.fedoraproject.org"
 id_provider_hostname = "id.fedoraproject.org"
 fasjson_base_url = "https://fasjson.fedoraproject.org"
 
+# Cache configuation
+[consumer_config.cache]
+backend = "dogpile.cache.memory"
+
 # This is a set of data that tells our consumer what Open Badges Issuer
 # should be kept as the issuer of all the badges we create.
 [consumer_config.badge_issuer]

fedbadges/cached.py

@@ -0,0 +1,62 @@
+import logging
+from contextlib import suppress
+
+import datanommer.models
+from dogpile.cache import make_region
+from dogpile.cache.api import NO_VALUE
+from fedora_messaging.message import Message
+
+
+log = logging.getLogger(__name__)
+cache = make_region()
+
+
+class CachedValue:
+
+    def _get_key(self, **kwargs):
+        return f"{self.__class__.__name__}:{kwargs!r}"
+
+    def get(self, **kwargs):
+        key = self._get_key(**kwargs)
+        return cache.get_or_create(key, creator=self.compute, creator_args=((), kwargs))
+
+    def set(self, *, value, **kwargs):
+        key = self._get_key(**kwargs)
+        cache.set(key, value)
+
+    def compute(self, **kwargs):
+        raise NotImplementedError
+
+    def on_message(self, message: Message):
+        raise NotImplementedError
+
+
+class TopicUserCount(CachedValue):
+
+    def compute(self, *, topic, username):
+        total, pages, query = datanommer.models.Message.grep(
+            topics=[topic], users=[username], defer=True
+        )
+        return total
+
+    def on_message(self, message: Message):
+        for username in message.usernames:
+            current_value = self.get(topic=message.topic, username=username)
+            if current_value == NO_VALUE:
+                continue  # Don't update the value if no one has ever requested it
+            self.set(self, topic=message.topic, username=username, value=current_value + 1)
+
+    @classmethod
+    def is_applicable(cls, search_kwargs):
+        """Return whether we can use this cached value for this datanommer query"""
+        query_keys = set(search_kwargs.keys())
+        with suppress(KeyError):
+            query_keys.remove("rows_per_page")
+        if query_keys != {"topics", "users"}:
+            return False
+        return len(search_kwargs["topics"]) == 1 and len(search_kwargs["users"]) == 1
+
+
+def on_message(message: Message):
+    log.debug("Updating cached values for message %s on %s", message.id, message.topic)
+    TopicUserCount().on_message(message)

fedbadges/consumer.py

@@ -19,6 +19,8 @@
 from fedora_messaging.config import conf as fm_config
 
 from .aio import Periodic
+from .cached import cache
+from .cached import on_message as update_cache_on_message
 from .rulesrepo import RulesRepo
 from .utils import datanommer_has_message, notification_callback
 
@@ -50,6 +52,8 @@ async def setup(self):
         # 1) Initialize our connection to the Tahrir DB
         # 2) Initialize our connection to the datanommer DB.
         # 3) Load our badge definitions and rules from YAML.
+        # Cache
+        await self.loop.run_in_executor(None, self._initialize_cache)
 
         # Tahrir stuff.
         await self.loop.run_in_executor(None, self._initialize_tahrir_connection)
@@ -71,6 +75,10 @@ async def setup(self):
         )
         await self._refresh_badges_task.start(run_now=True)
 
+    def _initialize_cache(self):
+        cache_args = self.config.get("cache")
+        cache.configure(**cache_args)
+
     def _initialize_tahrir_connection(self):
         database_uri = self.config.get("database_uri")
         if not database_uri:
@@ -128,6 +136,9 @@ def __call__(self, message: Message):
 
         # Award every badge as appropriate.
         log.debug("Received %s, %s", message.topic, message.id)
+
+        update_cache_on_message(message)
+
         tahrir = self._get_tahrir_client()
         for badge_rule in self.badge_rules:
             try:

fedbadges/rules.py

@@ -17,6 +17,7 @@
 from fedora_messaging.api import Message
 from tahrir_api.dbapi import TahrirDatabase
 
+from fedbadges.cached import TopicUserCount
 from fedbadges.utils import (
     # These are all in-process utilities
     construct_substitutions,
@@ -467,6 +468,12 @@ def _construct_query(self, msg):
             users = get_pagure_authors(kwargs["users"])
             if users:
                 kwargs["users"] = users
+
+        if TopicUserCount.is_applicable(kwargs) and self._d["operation"] == "count":
+            cached_value = TopicUserCount()
+            total = cached_value.get(topic=kwargs["topics"][0], username=kwargs["users"][0])
+            return total, None, None
+
         log.debug("Making datanommer query: %r", kwargs)
         kwargs["defer"] = True
         total, pages, query = datanommer.models.Message.grep(**kwargs)

pyproject.toml

@@ -44,6 +44,8 @@ backoff = "^2.2.1"
 fasjson-client = "^1.0.8"
 click = "^8.0.0"
 tahrir-api = "^1.0.0"
+dogpile-cache = "^1.3.2"
+pymemcache = {version = "^4.0.0", optional = true}
 
 # Message schemas. The reference list of all message schemas is in
 # https://github.com/fedora-infra/fedora-messaging/blob/develop/docs/schema-packages.txt
@@ -112,6 +114,9 @@ schemas = [
   "pagure-messages",
   "tahrir-messages",
 ]
+memcache = [
+  "pymemcache",
+]
 
 [tool.poetry.scripts]
 award-badges-dev = "fedbadges.manual.badges_dev:main"