Merge pull request #117 from dabercro/oracle

Major updates

Author: Daniel Abercrombie

bin/workflowtool

@@ -22,14 +22,15 @@
         'cherrypy<18.0.0',
         'mako',
         'numpy>=1.6.1',
-        'scipy>=0.19.1',
+        'scipy==1.1.0',
         'sklearn',
         'passlib>=1.6',
         'bcrypt',
         'pyOpenSSL',
         'pyyaml',
         'validators',
         'tabulate',
-        'pymongo<3.5.0'
+        'pymongo<3.5.0',
+        'cx_Oracle'
         ]
     )

setup.py

@@ -4,6 +4,6 @@
 :author: Daniel Abercrombie <dabercro@mit.edu>
 """
 
-__version__ = '0.5.0'
+__version__ = '0.8.2'
 
 __all__ = []

workflowwebtools/__init__.py

@@ -79,6 +79,7 @@ def get_workflow_vectors(workflows, session=None, allmap=None):
         settings = serverconfig.config_dict()['cluster'][column]
         column_output[column] = [numpy.zeros(len(allmap[column])) for _ in workflows]
 
+        cherrypy.log('Getting db_lock: 3')
         curs.db_lock.acquire()
         curs.curs.execute("SELECT SUM(numbererrors), {0}, stepname "
                           "FROM workflows "
@@ -96,6 +97,7 @@ def get_workflow_vectors(workflows, session=None, allmap=None):
                     if stepname:
                         wfname = stepname.split('/')[1]
 
+        cherrypy.log('releasing db_lock: 3')
         curs.db_lock.release()
 
         # Preprocessing here

workflowwebtools/clusterworkflows.py

@@ -81,3 +81,4 @@ cluster:
 cache_refresh:
   errors: 345600
 workspace: '.'
+refresh_period: 15

workflowwebtools/default/config.yml

@@ -14,13 +14,33 @@
 
 import validators
 import cherrypy
+import cx_Oracle
 
 from cmstoolbox import sitereadiness
 from cmstoolbox.webtools import get_json
 
 from . import workflowinfo
 from . import serverconfig
 
+def errors_from_list(workflows):
+    """
+    :param list workflows: A list of workflows that are in assistance-manual
+    :returns: The errors for the workflows
+    :rtype: dict
+    """
+    indict = {}
+
+    for workflow in workflows:
+        base = workflowinfo.WorkflowInfo(workflow)
+        prep_id = base.get_prep_id()
+        for wkf in set(workflowinfo.PrepIDInfo(prep_id).get_workflows()):
+            indict.update(
+                workflowinfo.WorkflowInfo(wkf).get_errors(get_unreported=True)
+                )
+
+    return indict
+
+
 def open_location(data_location):
     """
     This function assumes that the contents of the location is in JSON format.
@@ -30,8 +50,19 @@ def open_location(data_location):
     :returns: information in the JSON file
     :rtype: dict
     """
+    config_dict = serverconfig.config_dict()
+
+    if 'oracle' in config_dict:
+        oracle_db_conn = cx_Oracle.connect(*config_dict['oracle']) # pylint:disable=c-extension-no-member
+        oracle_cursor = oracle_db_conn.cursor()
+        oracle_cursor.execute(
+            "SELECT NAME FROM CMS_UNIFIED_ADMIN.workflow WHERE lower(STATUS) LIKE '%manual%'")
+        wkfs = [row for row, in oracle_cursor]
+        oracle_db_conn.close()
+        cherrypy.log('Number of workflows from database: %i' % len(wkfs))
+        return errors_from_list(wkfs)
+
     raw = None
-    indict = {}
 
     if os.path.isfile(data_location):
         with open(data_location, 'r') as input_file:
@@ -41,7 +72,7 @@ def open_location(data_location):
         components = urlparse.urlparse(data_location)
 
         # Anything we need for the Shibboleth cookie could be in the config file
-        cookie_stuff = serverconfig.config_dict()['data']
+        cookie_stuff = config_dict['data']
 
         raw = get_json(components.netloc, components.path,
                        use_https=True,
@@ -56,16 +87,10 @@ def open_location(data_location):
     if not (keys and isinstance(raw[keys[0]], list)):
         return raw
 
-    for workflow, statuses in raw.iteritems():
-        if True in ['manual' in status for status in statuses]:
-            base = workflowinfo.WorkflowInfo(workflow)
-            prep_id = base.get_prep_id()
-            for wkf in set(workflowinfo.PrepIDInfo(prep_id).get_workflows()):
-                indict.update(
-                    workflowinfo.WorkflowInfo(wkf).get_errors(get_unreported=True)
-                    )
-
-    return indict
+    return errors_from_list([
+        workflow for workflow, statuses in raw.iteritems()
+        if True in ['manual' in status for status in statuses]
+    ])
 
 
 def get_list_info(status_list):

workflowwebtools/errorinfo.py

@@ -68,14 +68,19 @@ def execute(self, query, params=None):
         :rtype: list
         """
 
+        output = []
+
         self.db_lock.acquire()
-        if params:
-            self.curs.execute(query, params)
-        else:
-            self.curs.execute(query)
+        curs = self.conn.cursor()
+        try:
+            if params:
+                curs.execute(query, params)
+            else:
+                curs.execute(query)
 
-        output = list(self.curs.fetchall())
-        self.db_lock.release()
+            output = list(curs.fetchall())
+        finally:
+            self.db_lock.release()
 
         return output
 
@@ -258,11 +263,13 @@ def get_step_list(self, workflow):
 
         if self._step_list is None:
             self._step_list = defaultdict(list)
+            cherrypy.log('Getting db_lock: 2')
             self.db_lock.acquire()
             self.curs.execute('SELECT DISTINCT(stepname) FROM workflows ORDER BY stepname')
             for tup in self.curs.fetchall():
                 stepname = tup[0]
                 self._step_list[stepname.split('/')[1]].append(stepname)
+            cherrypy.log('Releasing db_lock: 2')
             self.db_lock.release()
 
         return self._step_list[workflow]
@@ -324,32 +331,39 @@ def check_session(session, can_refresh=False):
     """
 
     if session:
+        cherrypy.log('Getting global lock: 1')
         GLOBAL_LOCK.acquire()
         if session.get('lock') is None:
             session['lock'] = threading.Lock()
+        cherrypy.log('Releasing global lock: 1')
         GLOBAL_LOCK.release()
 
+        cherrypy.log('Getting session lock')
         session['lock'].acquire()
 
         if session.get('info') is None:
             session['info'] = ErrorInfo()
         theinfo = session.get('info')
 
     else:
+        cherrypy.log('Getting global lock: 2')
         GLOBAL_LOCK.acquire()
         global GLOBAL_INFO
         if GLOBAL_INFO is None:
             GLOBAL_INFO = ErrorInfo()
 
         theinfo = GLOBAL_INFO
+        cherrypy.log('Releasing global lock: 2')
         GLOBAL_LOCK.release()
 
     # If session ErrorInfo is old, set up another connection
-    if can_refresh and theinfo.timestamp < time.time() - 60*30:
+    if can_refresh and theinfo.timestamp < time.time() - \
+            60*serverconfig.config_dict()['refresh_period']:
         theinfo.teardown()
         theinfo.setup()
 
     if session:
+        cherrypy.log('Releasing session lock')
         session['lock'].release()
 
     return theinfo

workflowwebtools/errorutils.py

@@ -180,6 +180,23 @@ def submitaction(user, workflows, action, session=None, **kwargs):
     return workflows, reasons, params
 
 
+def submit2(documents): # pylint: disable=missing-docstring
+    coll = get_actions_collection()
+
+    for document in documents:
+        workflow = document['workflow']
+        params = document['parameters']
+
+        cherrypy.log('About to insert workflow: %s action: %s' % (workflow, params))
+
+        coll.update_one({'workflow': workflow},
+                        {'$set':
+                             {'timestamp': int(time.time()),
+                              'parameters': params,
+                              'acted': 0}},
+                        upsert=True)
+
+
 def get_actions(num_days=None, num_hours=24, acted=0):
     """Get the recent actions to be acted on in dictionary form