Merge remote branch 'origin/master' into user

Conflicts:
	workflowmonit/requirements.txt
	workflowmonit/stompAMQ.py

Author: Daniel Abercrombie

.travis.yml

@@ -11,7 +11,7 @@ services:
   - mongodb
 env: TMPDIR=$PWD/tmp
 install:
-  - python setup.py install
+  - pip install .
 script:
   - pip list
   - package=workflowwebtools opsspace-test

setup.py

@@ -19,6 +19,7 @@
         },
     install_requires=[
         'cmstoolbox>=0.12.0',
+        'CMSMonitoring',
         'more-itertools<6.0.0',
         'cherrypy<18.0.0',
         'mako',

workflowmonit/README.rst

@@ -1,14 +1,13 @@
 workflowmonit
 -------------
 
-Component which periodically fetches information of workflows in the system(list extracted from Unified database), process and organizes into static documents, then sends to CMSMONIT service via :py:mod:`workflowmonit.stompAMQ` for storage, monitoring and post-aggregation.
+Component which periodically fetches information of workflows in the system(list extracted from Unified database), process and organizes into static documents, then sends to CMSMONIT service via :py:mod:`CMSMonitoring.StompAMQ` for storage, monitoring and post-aggregation.
 
 - :ref:`usedApi-ref`
 - Composition
     - :ref:`wmCollector-ref`
     - :ref:`wmSender-ref`
     - :ref:`wmScheduler-ref`
-    - :ref:`wmStompAMQ-ref`
 - :ref:`wmDocExample-ref`
 
 .. _usedApi-ref:
@@ -55,20 +54,11 @@ Schedule the ``main`` function of :ref:`wmSender-ref` every hour with :py:mod:`s
         time.sleep(1)
 
 
-.. _wmStompAMQ-ref:
-
-stompAMQ
-~~~~~~~~
-
-.. automodule:: workflowmonit.stompAMQ
-   :members:
-
-
 .. _wmDocExample-ref:
 
 Document examples
 ~~~~~~~~~~~~~~~~~
 
 1. document describling a single workflow `example1 <http://wsi.web.cern.ch/wsi/public/toSaveExample4.json>`_
-2. document wrapped by ``stompAMQ`` as a batch `example2.1 <http://wsi.web.cern.ch/wsi/public/godummy2.json>`_, `example2.2 <http://wsi.web.cern.ch/wsi/public/amqMsg_190208-201142.json>`_
-3. document sent out by ``stomp`` (what ``stompAMQ`` wrapped around) `example3 <http://wsi.web.cern.ch/wsi/public/bab2ef60-b0f2-4b55-9434-95a9cfd00510.json>`_
+2. document wrapped by ``StompAMQ`` as a batch `example2.1 <http://wsi.web.cern.ch/wsi/public/godummy2.json>`_, `example2.2 <http://wsi.web.cern.ch/wsi/public/amqMsg_190208-201142.json>`_
+3. document sent out by ``stomp`` (what ``StompAMQ`` wrapped around) `example3 <http://wsi.web.cern.ch/wsi/public/bab2ef60-b0f2-4b55-9434-95a9cfd00510.json>`_

workflowmonit/alertingDefs.py

@@ -51,7 +51,7 @@ def alertWithEmail(docs, recipients):
 
     for doc in docs:
         alertResults = [ad(doc) for ad in AlertDefs]
-        positiveRes = filter(lambda d: d[0], alertResults)
+        positiveRes = [r for r in alertResults if r[0]]
         if positiveRes:
             shortAlertMsgs = [x[1] for x in positiveRes]
             _contentMsg = '\n\n'.join([

workflowmonit/dumpWfStatusDb.py

@@ -21,7 +21,7 @@ def main():
     conn = sqlite3.connect(dbPath)
     with conn:
         c = conn.cursor()
-        for row in c.execute("SELECT * FROM workflowStatuses WHERE status='running-open' ORDER BY failurerate"):
+        for row in c.execute("SELECT * FROM workflowStatuses WHERE status LIKE '%archived' ORDER BY failurerate"):
             print('[ {1:^15} ]\t{2:.6f}\t{0}'.format(*row))
 
 if __name__ == "__main__":

workflowmonit/requirements.txt

@@ -1,4 +1,2 @@
 schedule
-CMSMonitoring
-stomp.py
 pyyaml>=5.1

workflowmonit/sendToMonit.py

@@ -8,10 +8,13 @@
 import logging
 import threading
 import logging.config
-from Queue import Queue
+try:
+    from Queue import Queue
+except ImportError:
+    from queue import Queue # pylint: disable=import-error
 
 import yaml
-from workflowmonit.stompAMQ import stompAMQ
+from CMSMonitoring.StompAMQ import StompAMQ
 import workflowmonit.workflowCollector as wc
 import workflowmonit.alertingDefs as ad
 
@@ -99,14 +102,7 @@ def getCompletedWorkflowsFromDb(configPath):
     """
     Get completed workflow list from local status db (setup to avoid unnecessary caching)
 
-    Workflows whose status is one of
-
-    - *running-closed*
-    - *completed*
-    - *aborted-archived*
-    - *rejected-arcived*
-
-    are removed from further caching.
+    Workflows whose status ends with *archived* are removed from further caching.
 
     :param str configPath: location of config file
     :returns: list of workflow (str)
@@ -127,7 +123,7 @@ def getCompletedWorkflowsFromDb(configPath):
         status TEXT,
         failurerate REAL
     );"""
-    DB_QUERY_CMD = """SELECT * FROM workflowStatuses WHERE status IN ('running-closed', 'completed', 'aborted-archived', 'rejected-archived')"""
+    DB_QUERY_CMD = """SELECT * FROM workflowStatuses WHERE status LIKE '%archived'"""
 
     res = []
     conn = sqlite3.connect(dbPath)
@@ -205,20 +201,21 @@ def buildDoc(configpath):
 
     wc.invalidate_caches('/tmp/wsi/workflowinfo')
 
-    q = TimeoutQueue()
-    num_threads = min(150, len(wkfs))
-    for wf in wkfs:
-        q.put(wf)
-
     results = list()
-    for _ in range(num_threads):
-        t = threading.Thread(target=worker, args=(results, q, completedWfs, ))
-        t.daemon = True
-        t.start()
-    try:
-        q.join_with_timeout(30*60)  # timeout 30min
-    except NotFinished:
-        pass
+
+    q = TimeoutQueue(maxsize=500)
+    num_threads = 500
+    for i, wf in enumerate(wkfs, 1):
+        q.put(wf)
+        if q.full() or i==len(wkfs):
+            for _ in range(num_threads):
+                t = threading.Thread(target=worker, args=(results, q, completedWfs, ))
+                t.daemon = True
+                t.start()
+            try:
+                q.join_with_timeout(30*60)  # timeout 30min
+            except NotFinished:
+                pass
 
     updateWorkflowStatusToDb(configpath, results)
     logger.info('Number of updated workflows: {}'.format(len(results)))
@@ -240,12 +237,12 @@ def sendDoc(cred, docs):
         return []
 
     try:
-        amq = stompAMQ(
-            None,  # username
-            None,  # password
-            cred['producer'],
-            cred['topic'],
-            # default [('agileinf-mb.cern.ch', 61213)]
+        amq = StompAMQ(
+            username = None,
+            password = None,
+            producer = cred['producer'],
+            topic = cred['topic'],
+            validation_schema = None,
             host_and_ports=[
                 (cred['hostport']['host'], cred['hostport']['port'])],
             logger=logger,
@@ -255,7 +252,7 @@ def sendDoc(cred, docs):
 
         doctype = 'workflowmonit_{}'.format(cred['producer'])
         notifications = [amq.make_notification(
-            payload=doc, docType=doctype) for doc in docs]
+            payload=doc, docType=doctype)[0] for doc in docs]
         failures = amq.send(notifications)
 
         logger.info("{}/{} docs successfully sent to AMQ.".format(
@@ -265,6 +262,7 @@ def sendDoc(cred, docs):
     except Exception as e:
         logger.exception(
             "Failed to send data to StompAMQ. Error: {}".format(str(e)))
+        raise
 
 
 def main():
@@ -291,17 +289,18 @@ def main():
 
         doc_bkp = os.path.join(LOGDIR, 'toSendDoc_{}'.format(
             time.strftime('%y%m%d-%H%M%S')))
-        wc.save_json(docs, doc_bkp)
-        logger.info('Document saved at: {}.json'.format(doc_bkp))
+        docfn = wc.save_json(docs, filename=doc_bkp, gzipped=True)
+        logger.info('Document saved at: {}'.format(docfn))
 
         failures = sendDoc(cred=cred, docs=docs)
 
         failedDocs_bkp = os.path.join(
             LOGDIR, 'amqFailedMsg_{}'.format(time.strftime('%y%m%d-%H%M%S')))
         if len(failures):
-            wc.save_json(failures, failedDocs_bkp)
-            logger.info('Failed message saved at: {}.json'.format(failedDocs_bkp))
+            failedDocFn = wc.save_json(failures, filename=failedDocs_bkp, gzipped=True)
+            logger.info('Failed message saved at: {}'.format(failedDocFn))
     except Exception as e:
+        logger.exception("Exception encounted, sending emails to {}".format(str(recipients)))
         ad.errorEmailShooter(str(e), recipients)
 
 

workflowmonit/stompAMQ.py

@@ -6,6 +6,7 @@
 import sys
 import json
 import time
+import gzip
 import shutil
 import threading
 from collections import defaultdict
@@ -16,18 +17,30 @@
 from workflowwebtools import errorutils
 
 
-def save_json(json_obj, filename='tmp'):
+def save_json(json_obj, filename='tmp', gzipped=False):
     """
     save json object to a local formatted text file, for debug
 
     :param dict json_obj: the json object
     :param str filename: the base name of the file to be saved
-    :returns: None
+    :param bool gzipped: if gzip output document, default is False
+    :returns: full filename
+
+    :rtype: str
     """
 
-    with open('%s.json' % filename, 'w') as tmp:
-        tmp.write(json.dumps(json_obj, sort_keys=True,
-                             indent=4, separators=(',', ': ')))
+    fn = "{}.json".format(filename)
+    msg = json.dumps(json_obj, sort_keys=True, indent=4, separators=(',', ': '))
+
+    if gzipped:
+        fn += '.gz'
+        with gzip.open(fn, 'wb') as f:
+            f.write(msg.encode())
+    else:
+        with open(fn, 'w') as f:
+            f.write(msg)
+
+    return fn
 
 
 def get_yamlconfig(configPath):
@@ -619,7 +632,10 @@ def main():
     print("Number of workflows retrieved from Oracle DB: ", len(wfs))
     invalidate_caches()
 
-    from Queue import Queue
+    try:
+        from Queue import Queue
+    except ImportError:
+        from queue import Queue # pylint: disable=import-error
     q = Queue()
     num_threads = min(150, len(wfs))
 

workflowmonit/workflowCollector.py

@@ -10,7 +10,7 @@
 LOGDIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'Logs')
 LOGGING_CONFIG = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'configLogging.yml')
 
-schedule.every().hour.do(wms.main)
+schedule.every(30).minutes.do(wms.main)
 
 
 if __name__ == "__main__":