Improve performance in a single run

pacifica/elasticsearch/render/base.py

@@ -5,7 +5,7 @@
 from pacifica.metadata.rest.objectinfo import ObjectInfoAPI
 from pacifica.metadata.orm import Relationships
 from ..config import get_config
-
+from memoization import cached
 
 _LRU_GLOBAL_ARGS = {
     'maxsize': get_config().getint('elasticsearch', 'cache_size'),
@@ -46,6 +46,18 @@ class SearchBase:
     releaser_uuid = str(Relationships.get(Relationships.name == 'authorized_releaser').uuid)
     search_required_uuid = str(Relationships.get(Relationships.name == 'search_required').uuid)
 
+
+    @classmethod
+    @query_select_default_args
+    def get_index_query(cls,obj_cls,**kwargs):
+        """Generate the select query to give all the rows of class"""
+        return (obj_cls.select(obj_cls.id))  # this shoudl work, but peewee borks it
+        #return (obj_cls.select())
+
+    @classmethod
+    def get_render_query(cls,obj_cls,id):
+        return (obj_cls.select().where(obj_cls.id == id))
+
     @classmethod
     @query_select_default_args
     def get_select_query(cls, time_delta, obj_cls, time_field):
@@ -82,6 +94,7 @@ def _cls_name_to_module(rel_cls):
         return rel_cls.__module__.split('.')[-1]
 
     @classmethod
+    @cached(max_size=5000)
     def render(cls, obj, render_rel_objs=False, render_trans_ids=False):
         """Render the object and return it."""
         ret = {'type': cls._cls_name_to_module(cls)}

pacifica/elasticsearch/render/groups.py

@@ -20,6 +20,16 @@ class GroupsRender(SearchBase):
         'instruments'
     ]
 
+    @classmethod
+    def get_render_query(cls,obj_cls,id):
+        """Generate the select query for groups related to instruments."""
+        return (
+            Groups.select()
+            .join(InstrumentGroup, JOIN.LEFT_OUTER, on=(InstrumentGroup.group == Groups.id))
+            .join(Instruments, JOIN.LEFT_OUTER, on=(Instruments.id == InstrumentGroup.instrument))
+            .where( Groups.id == id )
+        )
+
     @classmethod
     @query_select_default_args
     def get_select_query(cls, time_delta, obj_cls, time_field):

pacifica/elasticsearch/render/institutions.py

@@ -16,6 +16,25 @@ class InstitutionsRender(SearchBase):
         'updated_date', 'created_date'
     ]
 
+    @classmethod
+    def get_render_query(cls,obj_cls,id):
+        """Generate the select query for groups related to instruments."""
+        # pylint: disable=invalid-name
+        SIPTrans = Transactions.alias()
+        SAPTrans = Transactions.alias()
+        # pylint: enable=invalid-name
+        return (
+            Institutions.select()
+            .join(InstitutionUser, JOIN.LEFT_OUTER, on=(InstitutionUser.institution == Institutions.id))
+            .join(Users, JOIN.LEFT_OUTER, on=(InstitutionUser.user == Users.id))
+            .join(TransSIP, JOIN.LEFT_OUTER, on=(TransSIP.submitter == Users.id))
+            .join(SIPTrans, JOIN.LEFT_OUTER, on=(SIPTrans.id == TransSIP.id))
+            .join(TransSAP, JOIN.LEFT_OUTER, on=(TransSAP.submitter == Users.id))
+            .join(SAPTrans, JOIN.LEFT_OUTER, on=(SAPTrans.id == TransSAP.id))
+            .where( Institutions.id == id )
+        )
+
+
     @classmethod
     @query_select_default_args
     def get_select_query(cls, time_delta, obj_cls, time_field):

pacifica/elasticsearch/render/instruments.py

@@ -18,6 +18,17 @@ class InstrumentsRender(SearchBase):
     ]
     rel_objs = ['key_value_pairs']
 
+    @classmethod
+    def get_render_query(cls,obj_cls,id):
+        return (
+            Instruments.select()
+            .join(InstrumentKeyValue, JOIN.LEFT_OUTER, on=(InstrumentKeyValue.instrument == Instruments.id))
+            .join(Keys, JOIN.LEFT_OUTER, on=(InstrumentKeyValue.key == Keys.id))
+            .join(Values, JOIN.LEFT_OUTER, on=(InstrumentKeyValue.key == Values.id))
+            .join(TransSIP, JOIN.LEFT_OUTER, on=(TransSIP.instrument == Instruments.id))
+            .where(Instruments.id == id)
+        )
+
     @classmethod
     @query_select_default_args
     def get_select_query(cls, time_delta, obj_cls, time_field):

pacifica/elasticsearch/render/projects.py

@@ -27,6 +27,24 @@ class ProjectsRender(SearchBase):
         'released_count', 'science_themes'
     ]
 
+    @classmethod
+    def get_render_query(cls,obj_cls,id):
+        return (
+            Projects.select()
+            .join(ProjectUser, JOIN.LEFT_OUTER, on=(ProjectUser.project == Projects.id))
+            .join(Users, JOIN.LEFT_OUTER, on=(ProjectUser.user == Users.id))
+            .join(Relationships, JOIN.LEFT_OUTER, on=(Relationships.uuid == ProjectUser.relationship))
+            .join(InstitutionUser, JOIN.LEFT_OUTER, on=(Users.id == InstitutionUser.user))
+            .join(Institutions, JOIN.LEFT_OUTER, on=(InstitutionUser.institution == Institutions.id))
+            .join(ProjectInstrument, JOIN.LEFT_OUTER, on=(ProjectInstrument.project == Projects.id))
+            .join(Instruments, JOIN.LEFT_OUTER, on=(ProjectInstrument.instrument == Instruments.id))
+            .join(InstrumentGroup, JOIN.LEFT_OUTER, on=(InstrumentGroup.instrument == Instruments.id))
+            .join(Groups, JOIN.LEFT_OUTER, on=(InstrumentGroup.group == Groups.id))
+            .join(TransSIP, JOIN.LEFT_OUTER, on=(TransSIP.project == Projects.id))
+            .join(TransSAP, JOIN.LEFT_OUTER, on=(TransSAP.project == Projects.id))
+            .where(Projects.id == id)
+        )
+
     @classmethod
     @query_select_default_args
     def get_select_query(cls, time_delta, obj_cls, time_field):
@@ -198,8 +216,11 @@ def users_obj_lists(cls, **proj_obj):
         for proj_user_obj in cls.get_rel_by_args('project_user', project=proj_obj['_id']):
             rel_obj = cls.get_rel_by_args('relationships', uuid=proj_user_obj['relationship'])[0]
             rel_list = ret.get(rel_obj['name'], [])
-            rel_list.append(
-                UsersRender.render(cls.get_rel_by_args('users', _id=proj_user_obj['user'])[0])
-            )
+            try:
+                rel_list.append(
+                    UsersRender.render(cls.get_rel_by_args('users', _id=proj_user_obj['user'])[0])
+                )
+            except IndexError:
+                print("IndexError getting user for project ", proj_user_obj)
             ret[rel_obj['name']] = rel_list
         return ret

pacifica/elasticsearch/render/relationships.py

@@ -13,6 +13,15 @@ class RelationshipsRender(SearchBase):
         'updated_date', 'created_date'
     ]
 
+    @classmethod
+    def get_index_query(cls,obj_cls,**kwargs):
+        """Generate the select query to give all the rows of class"""
+        return (obj_cls.select(obj_cls.uuid))
+
+    @classmethod
+    def get_render_query(cls,obj_cls,id):
+        return (obj_cls.select().where(obj_cls.uuid == id))
+
     @staticmethod
     def obj_id(**rel_obj):
         """Return string for object id."""

pacifica/elasticsearch/render/science_themes.py

@@ -3,7 +3,7 @@
 """Search transaction rendering methods."""
 from six import text_type
 from .base import SearchBase
-
+from peewee import DatabaseError
 
 class ScienceThemesRender(SearchBase):
     """Render an science theme for search."""
@@ -48,8 +48,11 @@ def get_transactions(cls, **proj_obj):
         """Return the list of transaction ids for the science theme."""
         ret = set()
         for rel_proj_obj in cls.get_rel_by_args('projects', science_theme=proj_obj['science_theme']):
-            ret.update([
-                'transactions_{}'.format(trans_id)
-                for trans_id in cls._transsip_transsap_merge({'project': rel_proj_obj['_id']}, '_id')
-            ])
+            try:
+                ret.update([
+                    'transactions_{}'.format(trans_id)
+                    for trans_id in cls._transsip_transsap_merge({'project': rel_proj_obj['_id']}, '_id')
+                ])
+            except peewee.DatabaseError as msg:
+                print("DatabaseError rendering transactions for science themes ",proj_obj,msg)
         return list(ret)

pacifica/elasticsearch/render/transactions.py

@@ -27,6 +27,38 @@ class TransactionsRender(SearchBase):
     rel_objs = [
         'users', 'instruments', 'groups', 'projects', 'key_value_pairs', 'files'
     ]
+    @classmethod
+    def get_render_query(cls,obj_cls,id):
+        # The alias() method does return a class
+        # pylint: disable=invalid-name
+        ReleaseUsers = Users.alias()
+        SIPUsers = Users.alias()
+        SAPUsers = Users.alias()
+        SIPProjects = Projects.alias()
+        SAPProjects = Projects.alias()
+        # pylint: enable=invalid-name
+        return (
+            Transactions.select()
+            .join(TransactionUser, JOIN.LEFT_OUTER, on=(TransactionUser.transaction == Transactions.id))
+            .join(DOITransaction, JOIN.LEFT_OUTER, on=(DOITransaction.transaction == TransactionUser.uuid))
+            .join(Relationships, JOIN.LEFT_OUTER, on=(Relationships.uuid == TransactionUser.relationship))
+            .join(ReleaseUsers, JOIN.LEFT_OUTER, on=(TransactionUser.user == ReleaseUsers.id))
+            .join(TransSIP, JOIN.LEFT_OUTER, on=(TransSIP.id == Transactions.id))
+            .join(TransSAP, JOIN.LEFT_OUTER, on=(TransSAP.id == Transactions.id))
+            .join(SIPUsers, JOIN.LEFT_OUTER, on=(TransSIP.submitter == SIPUsers.id))
+            .join(SAPUsers, JOIN.LEFT_OUTER, on=(TransSAP.submitter == SAPUsers.id))
+            .join(Instruments, JOIN.LEFT_OUTER, on=(Instruments.id == TransSIP.instrument))
+            .join(SIPProjects, JOIN.LEFT_OUTER, on=(SIPProjects.id == TransSIP.project))
+            .join(SAPProjects, JOIN.LEFT_OUTER, on=(SAPProjects.id == TransSAP.project))
+            .join(InstrumentGroup, JOIN.LEFT_OUTER, on=(InstrumentGroup.instrument == TransSIP.instrument))
+            .join(Groups, JOIN.LEFT_OUTER, on=(Groups.id == InstrumentGroup.group))
+            .join(Files, JOIN.LEFT_OUTER, on=(Files.transaction == Transactions.id))
+            .join(TransactionKeyValue, JOIN.LEFT_OUTER, on=(TransactionKeyValue.transaction == Transactions.id))
+            .join(Keys, JOIN.LEFT_OUTER, on=(TransactionKeyValue.key == Keys.id))
+            .join(Values, JOIN.LEFT_OUTER, on=(TransactionKeyValue.value == Values.id))
+            .where(Transactions.id==id)
+        )
+
 
     @classmethod
     @query_select_default_args
@@ -160,9 +192,12 @@ def users_obj_lists(cls, **trans_obj):
         """Get the user objects and relationships."""
         ret = {'submitter': []}
         for user_id in cls._transsip_transsap_merge({'_id': trans_obj['_id']}, 'submitter'):
-            ret['submitter'].append(
-                UsersRender.render(cls.get_rel_by_args('users', _id=user_id)[0])
-            )
+            try:
+                ret['submitter'].append(
+                    UsersRender.render(cls.get_rel_by_args('users', _id=user_id)[0])
+                )
+            except IndexError: # This happens when a user does not exist, or is 'deleted'
+                pass
         for trans_user_obj in cls.get_rel_by_args('transaction_user', transaction=trans_obj['_id']):
             rel_obj = cls.get_rel_by_args('relationships', uuid=trans_user_obj['relationship'])[0]
             rel_list = ret.get(rel_obj['name'], [])
@@ -200,12 +235,14 @@ def groups_obj_lists(cls, **trans_obj):
     @classmethod
     def projects_obj_lists(cls, **trans_obj):
         """Get the projects related to the transaction."""
-        return [
-            ProjectsRender.render(
-                cls.get_rel_by_args('projects', _id=proj_id)[0]
-            ) for proj_id in cls._transsip_transsap_merge({'_id': trans_obj['_id']}, 'project')
-        ]
-
+        try:
+            return [
+                ProjectsRender.render(
+                    cls.get_rel_by_args('projects', _id=proj_id)[0]
+                ) for proj_id in cls._transsip_transsap_merge({'_id': trans_obj['_id']}, 'project')
+            ]
+        except IndexError:
+            print("IndexError rendering project for transaction ",trans_obj)
     @classmethod
     def key_value_pairs_obj_lists(cls, **trans_obj):
         """Get the key value pairs related to the transaction."""

pacifica/elasticsearch/render/users.py

@@ -15,6 +15,15 @@ class UsersRender(SearchBase):
         'updated_date', 'created_date'
     ]
 
+    @classmethod
+    def get_render_query(cls,obj_cls,id):
+         return (
+            Users.select()
+            .join(TransSIP, JOIN.LEFT_OUTER, on=(TransSIP.submitter == Users.id))
+            .join(TransSAP, JOIN.LEFT_OUTER, on=(TransSAP.submitter == Users.id))
+            .where(Users.id==id)
+        )
+
     @classmethod
     @query_select_default_args
     def get_select_query(cls, time_delta, obj_cls, time_field):

pacifica/elasticsearch/search_render.py

@@ -32,10 +32,14 @@ def object_exclude(cls, obj_cls, obj, exclude):
         return False
 
     @classmethod
-    def generate(cls, obj_cls, objs, exclude):
+    def generate(cls, obj_cls, objs, exclude, db_cls):
         """generate the institution object."""
+        #print("generate",cls,obj_cls,objs,exclude)
         render_cls = cls.get_render_class(obj_cls)
-        for obj in objs:
+        #print(render_cls)
+        #convert to full renderable objects
+        renderable_objs = [render_cls.get_render_query(db_cls,r).get().to_hash() for r in objs]
+        for obj in renderable_objs:
             if cls.object_exclude(obj_cls, obj, exclude):
                 continue
             yield {

pacifica/elasticsearch/search_sync.py

@@ -5,8 +5,7 @@
 import os
 from json import dumps, loads
 from time import sleep
-from threading import Thread
-from queue import Queue
+from multiprocessing import Process, JoinableQueue
 from math import ceil
 from datetime import datetime
 from itertools import zip_longest
@@ -77,12 +76,18 @@ def try_doing_work(cli, job):
     """Try doing some work even if you fail."""
     tries_left = 5
     success = False
+    data = yield_data(**job)
     while not success and tries_left:
         try:
-            helpers.bulk(cli, yield_data(**job))
+            helpers.bulk(cli, data)
             success = True
-        except ElasticsearchException:  # pragma: no cover
+        except ElasticsearchException as msg:  # pragma: no cover
+            print("ElasticsearchException:",tries_left,msg[:100])
             tries_left -= 1
+        except MemoryError as msg: # pragma: no cover
+            print("MemoryError:",tries_left,msg[:100])
+            tries_left -= 1
+            sleep(5)
     return success
 
 
@@ -92,15 +97,15 @@ def yield_data(**kwargs):
     exclude = kwargs.pop('exclude')
     obj_cls = ObjectInfoAPI.get_class_object_from_name(obj)
     render_cls = SearchRender.get_render_class(obj)
-    query = render_cls.get_select_query(obj_cls=obj_cls, **kwargs)
-    return SearchRender.generate(obj, [qobj.to_hash() for qobj in query], exclude)
+    query = render_cls.get_index_query(obj_cls=obj_cls, **kwargs)
+    return SearchRender.generate(obj, [qobj for qobj in query], exclude, obj_cls)
 
 
 def create_worker_threads(threads, work_queue):
     """Create the worker threads and return the list."""
     work_threads = []
     for _i in range(threads):
-        wthread = Thread(target=start_work, args=(work_queue,))
+        wthread = Process(target=start_work, args=(work_queue,))
         wthread.daemon = True
         wthread.start()
         work_threads.append(wthread)
@@ -116,9 +121,9 @@ def generate_work(args, work_queue):
         for time_field in args.compare_dates:
             obj_cls = ObjectInfoAPI.get_class_object_from_name(obj)
             render_cls = SearchRender.get_render_class(obj)
-            query = render_cls.get_select_query(
-                obj_cls=obj_cls, time_delta=time_delta,
-                enable_paging=False, time_field=time_field
+            query = render_cls.get_index_query(
+                obj_cls=obj_cls,
+                enable_paging=False,
             )
             num_pages = int(ceil(float(query.count()) / args.items_per_page))
             for page in range(1, num_pages + 1):
@@ -144,7 +149,7 @@ def search_sync(args):
     if args.celery:
         work_queue = CeleryQueue()
     else:
-        work_queue = Queue(32)
+        work_queue = JoinableQueue(32)
         work_threads = create_worker_threads(args.threads, work_queue)
     generate_work(args, work_queue)
     if args.celery:

setup.py

@@ -25,7 +25,8 @@
         'pacifica-metadata>=0.11.0,<1',
         'pacifica-namespace',
         'python-dateutil',
-        'tqdm'
+        'tqdm',
+        'memoization'
     ],
     include_package_data=True,
     package_data={'': ['*.json']},