redis for solr core creation.

Author: JVickery-TBS

.gitignore

@@ -5,6 +5,7 @@ bower_components
 # Byte-compiled / optimized / DLL files
 __pycache__/
 *.py[cod]
+**/venv/**
 
 # C extensions
 *.so

ckanext/datastore_search/backend/solr.py

@@ -2,12 +2,14 @@
 import json
 import requests
 import re
+import rq
 from logging import getLogger
 
 from typing import Any, Optional, Dict, cast, List
 from ckan.types import Context, DataDict
 
 from ckan.plugins.toolkit import _, config, get_action
+from ckan.lib.redis import connect_to_redis
 
 from ckanext.datastore.logic.action import datastore_search_sql
 from ckanext.datastore.backend.postgres import identifier
@@ -20,10 +22,13 @@
 
 MAX_ERR_LEN = 1000
 PSQL_TO_SOLR_WILCARD_MATCH = re.compile('^_?|_?$')
+REDIS_QUEUE_NAME = 'ckan_ds_solr_core_create'
 
 log = getLogger(__name__)
 DEBUG = config.get('debug', False)
 
+_ds_solr_queues: Dict[str, rq.Queue] = {}
+
 
 class DatastoreSolrBackend(DatastoreSearchBackend):
     """
@@ -277,17 +282,28 @@ def create(self,
             #        solr create -c core_name -d configsets/datastore_resource
             #        then does SOLR server need to send a signal back?
             #        or can we just keep retrying a couple of times??
+            global _ds_solr_queues
+            redis_queue = None
             errmsg = _('Could not create SOLR core %s') % core_name
-            req_body = {'create': [{'name': core_name,
-                                    'configSet': 'datastore_resource'}]}
-            resp = self._send_api_request(method='POST',
-                                          endpoint='cores',
-                                          body=req_body)
-            if 'error' in resp:
-                raise DatastoreSearchException(
-                    errmsg if not DEBUG
-                    else resp['error'].get('msg', errmsg)[:MAX_ERR_LEN])
-            log.debug('Created SOLR Core for DataStore Resource %s' % resource_id)
+            if REDIS_QUEUE_NAME in _ds_solr_queues:
+                redis_queue = _ds_solr_queues[REDIS_QUEUE_NAME]
+            else:
+                redis_conn = connect_to_redis()
+                redis_queue = _ds_solr_queues[REDIS_QUEUE_NAME] = \
+                    rq.Queue(REDIS_QUEUE_NAME, connection=redis_conn)
+            if not redis_queue:
+                raise DatastoreSearchException(errmsg)
+            job = redis_queue.enqueue_call(
+                'create_solr_core.proc._create_solr_core',
+                args=[core_name, 'datastore_resource'],
+                timeout=60)
+            if not job.meta:
+                job.meta = {}
+            job.meta['title'] = 'SOLR Core creation %s' % core_name
+            job.save()
+            log.debug('Enqueued SOLR Core creation for DataStore Resource %s ' %
+                      resource_id)
+            # TODO: await or retry here???
             conn = self._make_connection(resource_id)
         if not conn:
             raise DatastoreSearchException(

ckanext/datastore_search/config/redis/create_solr_core/__init__.py

@@ -0,0 +1,15 @@
+# encoding: utf-8
+__all__ = ['proc', 'worker']
+import create_solr_core.proc as mod_proc
+import create_solr_core.worker as mod_worker
+
+proc = mod_proc
+worker = mod_worker
+
+# this is a namespace package
+try:
+    import pkg_resources
+    pkg_resources.declare_namespace(__name__)
+except ImportError:
+    import pkgutil
+    __path__ = pkgutil.extend_path(__path__, __name__)

ckanext/datastore_search/config/redis/create_solr_core/proc.py

@@ -0,0 +1,8 @@
+import subprocess
+
+def _create_solr_core(core_name, config_set):
+    subprocess.run(
+        ['solr', 'create', '-c', core_name, '-d', config_set],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        timeout=60)

ckanext/datastore_search/config/redis/create_solr_core/worker.py

@@ -0,0 +1,56 @@
+import rq
+from redis import Redis, ConnectionPool
+import click
+from logging import getLogger
+
+from typing import Dict, Optional, Tuple
+
+
+log = getLogger(__name__)
+
+_solr_queues: Dict[str, rq.Queue] = {}
+_redis_connection_pool = None
+
+
+@click.command(short_help="Work the given queues.")
+@click.option('-r', '--redis-url', required=True,
+              type=click.STRING, default=None,
+              help='The REDIS connection URI.')
+@click.option('-q', '--queues', default=None, required=True,
+              multiple=True, help='Queue names to work on.')
+@click.option('-b', '--burst', is_flag=True, default=False,
+              type=click.BOOL, help='Burst the worker.')
+@click.option('-v', '--verbose', is_flag=True, default=False,
+              type=click.BOOL, help='Increase verbosity.')
+def _worker(redis_url: Optional[str] = None,
+            queues: Optional[Tuple[str]] = None,
+            burst: bool = False,
+            verbose: bool = False):
+    """
+    Work the given queues.
+    """
+    global _solr_queues
+    global _redis_connection_pool
+
+    if _redis_connection_pool is None:
+        if verbose:
+            click.echo('Using Redis at %s' % redis_url)
+        _redis_connection_pool = ConnectionPool.from_url(redis_url)
+    redis_conn = Redis(connection_pool=_redis_connection_pool)
+
+    redis_queues = {}
+    for queue in queues:
+        if queue in _solr_queues:
+            redis_queue = _solr_queues[queue]
+        else:
+            redis_queue = _solr_queues[queue] = \
+                rq.Queue(queue, connection=redis_conn)
+        redis_queues[queue] = redis_queue
+
+    worker = rq.Worker(queues=redis_queues.values(),
+                       connection=redis_conn)
+    worker.work(burst=burst)
+
+
+if __name__ == '__main__':
+    _worker()

ckanext/datastore_search/config/redis/requirements.txt

@@ -0,0 +1,3 @@
+rq==2.0.0
+redis==5.2.0
+click==8.1.8

ckanext/datastore_search/config/redis/setup.cfg

@@ -0,0 +1,26 @@
+[metadata]
+name = create_solr_core
+version = 0.0.1
+description = Python module to create SOLR cores.
+url = https://github.com/JVickery-TBS/ckanext-datastore-search
+author = Jesse Vickery
+author_email = jesse.vickery@tbs-sct.gc.ca
+license = AGPL
+classifiers =
+            Development Status :: 4 - Beta
+            License :: OSI Approved :: GNU Affero General Public License v3 or later (AGPLv3+)
+            Programming Language :: Python :: 3.7
+            Programming Language :: Python :: 3.8
+            Programming Language :: Python :: 3.9
+            Programming Language :: Python :: 3.10
+keywords = CKAN Datastore SOLR search index indices
+
+[options]
+packages = find_namespace:
+package_dir =
+  =create_solr_core
+install_requires =
+include_package_data = True
+
+[options.packages.find]
+where=create_solr_core

ckanext/datastore_search/config/redis/setup.py

@@ -0,0 +1,4 @@
+# -*- coding: utf-8 -*-
+from setuptools import setup
+
+setup()

ckanext/datastore_search/logic/__init__.py

@@ -1,2 +1,4 @@
 pysolr==3.10.0
 requests==2.32.3
+rq==2.0.0
+redis==5.2.0