Add command to clear stale cacheops keys (#434)

* Add conj-reaper management command

* Move main command logic into a function and call it from the command

Rework #323 into a reusable function that can be wrapped in a Celery task by

* Rename back to reaper and set default values in logic

Co-Authored-By: Alexander Schepanovski <suor.web@gmail.com>

* Don't hardcode default DB alias

* Expose reap_conjs on the package's top level API

* Update documentation

* Remove f-string as we're still supporting Python 3.5

* Fix comments in documentation

* Pluralize management command

Co-authored-by: Nic Wolff <nwolff@hearst.com>
Co-authored-by: Alexander Schepanovski <suor.web@gmail.com>

Author: 3 people

README.rst

@@ -725,6 +725,32 @@ Here is a simple stats implementation:
 Cache invalidation signal is emitted after object, model or global invalidation passing ``sender`` and ``obj_dict`` args. Note that during normal operation cacheops only uses object invalidation, calling it once for each model create/delete and twice for update: passing old and new object dictionary.
 
 
+Memory usage cleanup
+--------------------
+
+In some cases, cacheops may leave some conjunction keys of expired cache keys in redis without being able
+to invalidate them. Cacheops ships with a ``cacheops.reap_conjs`` function that can clean up these keys,
+ignoring conjunction sets with some reasonable size.
+
+It can be called using the ``reapconjs`` management command::
+
+    ./manage.py reapconjs --chunk-size=100 --min-conj-set-size=10000  # with custom values
+    ./manage.py reapconjs                                             # with default values (chunks=1000, min size=1000)
+
+The command is a small wrapper that calls a function with the main logic. You can also call it from your code, for example from a Celery task:
+
+.. code:: python
+
+    from cacheops import reap_conjs
+
+    @app.task
+    def reap_conjs_task():
+        reap_conjs(
+            chunk_size=2000,
+            min_conj_set_size=100,
+        )
+
+
 CAVEATS
 -------
 

cacheops/__init__.py

@@ -5,6 +5,7 @@
 from .simple import *  # noqa
 from .query import *  # noqa
 from .invalidation import *  # noqa
+from .reaper import *  # noqa
 from .templatetags.cacheops import *  # noqa
 
 

cacheops/management/commands/reapconjs.py

@@ -0,0 +1,21 @@
+from argparse import ArgumentParser
+
+from django.core.management.base import BaseCommand
+
+from cacheops.reaper import reap_conjs
+
+
+class Command(BaseCommand):
+    help = 'Removes expired conjunction keys from cacheops.'
+
+    def add_arguments(self, parser: ArgumentParser):
+        parser.add_argument('--chunk-size', type=int, default=1000)
+        parser.add_argument('--min-conj-set-size', type=int, default=1000)
+        parser.add_argument('--dry-run', action='store_true')
+
+    def handle(self, chunk_size: int, min_conj_set_size: int, dry_run: bool, **kwargs):
+        reap_conjs(
+            chunk_size=chunk_size,
+            min_conj_set_size=min_conj_set_size,
+            dry_run=dry_run,
+        )

cacheops/reaper.py

@@ -0,0 +1,65 @@
+import logging
+
+from django.db import DEFAULT_DB_ALIAS
+
+from .redis import redis_client
+from .sharding import get_prefix
+
+logger = logging.getLogger(__name__)
+
+
+def reap_conjs(
+    chunk_size: int = 1000,
+    min_conj_set_size: int = 1000,
+    using=DEFAULT_DB_ALIAS,
+    dry_run: bool = False,
+):
+    """
+    Remove expired cache keys from invalidation sets.
+
+    Cacheops saves each DB resultset cache key in a "conj set" so it can delete it later if it
+    thinks it should be invalidated due to a saved record with matching values. But the resultset
+    caches time out after 30 minutes, and their cache keys live in those conj sets forever!
+
+    So conj sets for frequent queries on tables that aren't updated often end up containing
+    millions of already-expired cache keys and maybe a few thousand actually useful ones,
+    and block Redis for multiple - or many - seconds when cacheops finally decides
+    to invalidate them.
+
+    This function scans cacheops' conj keys for already-expired cache keys and removes them.
+    """
+    logger.info('Starting scan for large conj sets')
+    prefix = get_prefix(dbs=[using])
+    for conj_key in redis_client.scan_iter(prefix + 'conj:*', count=chunk_size):
+        total = redis_client.scard(conj_key)
+        if total < min_conj_set_size:
+            continue
+        logger.info('Found %s cache keys in %s, scanning for expired keys', total, conj_key)
+        _clear_conj_key(conj_key, chunk_size, dry_run)
+    logger.info('Done scan for large conj sets')
+
+
+def _clear_conj_key(conj_key: bytes, chunk_size: int, dry_run: bool):
+    """Scan the cache keys in a conj set in batches and remove any that have expired."""
+    count, removed = 0, 0
+    for keys in _iter_keys_chunk(chunk_size, conj_key):
+        count += len(keys)
+        values = redis_client.mget(keys)
+        expired = [k for k, v in zip(keys, values) if not v]
+        if expired:
+            if not dry_run:
+                redis_client.srem(conj_key, *expired)
+            removed += len(expired)
+            logger.info('Removed %s/%s cache keys from %s', removed, count, conj_key)
+    if removed and not dry_run:
+        redis_client.execute_command('MEMORY PURGE')
+
+
+def _iter_keys_chunk(chunk_size, key):
+    cursor = 0
+    while True:
+        cursor, items = redis_client.sscan(key, cursor, count=chunk_size)
+        if items:
+            yield items
+        if cursor == 0:
+            break