Merge pull request #132 from phylsix/master

some bug fixes on /workflowmonit

Author: Daniel Abercrombie

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
-git+https://github.com/dmwm/WMCore.git
-stomp.py
 pyyaml>=5.1

workflowmonit/sendToMonit.py

@@ -8,7 +8,10 @@
 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 CMSMonitoring.StompAMQ import StompAMQ
@@ -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)))
@@ -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/workflowCollector.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/workflowMonitScheduler.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__":