Merge pull request #129 from phylsix/master

py3 for workflowmonit subdir

Author: Daniel Abercrombie

workflowmonit/alertingDefs.py

@@ -0,0 +1,112 @@
+#!/usr/bin/env python
+from __future__ import print_function
+import time
+import json
+import smtplib
+from email.mime.text import MIMEText
+
+
+def onFailureRate(doc, thres=0.5):
+    """
+    check a workflow (represented by `doc`), if its failureRate is larger than
+    0.5 AND running time is > 2days, then ALERT.
+
+    :param doc dict: information describe a workflow
+    :param thres float: threshold
+    :returns: (judge result, short msg if failed)
+    """
+
+    res = (False, '')
+    if doc['status'] != 'running-open': return res
+    if doc['failureRate'] < thres: return res
+    runningOpen = [
+        tr for tr in doc['transitions'] if tr['Status'] == 'running-open'
+    ]
+    if not runningOpen: return res
+
+    runningOpenTime = runningOpen[0]['UpdateTime']
+    if time.time() - runningOpenTime < 2 * 24 * 60 * 60: return res
+
+    failMsg = 'FailureRate ({}) larger than threshold({}), while running time over 2 days (started at {})'.format(
+        doc['failureRate'], thres, time.ctime(runningOpenTime))
+
+    return (True, failMsg)
+
+
+AlertDefs = [
+    onFailureRate,
+]
+
+
+def alertWithEmail(docs, recipients):
+    """
+    handling docs with alert emails.
+
+
+    :param docs list: list of documents
+    :param recipients list: list of recipients email addresses
+    """
+
+    sender = 'toolsandint-workflowmonitalert@cern.ch'
+
+    for doc in docs:
+        alertResults = [ad(doc) for ad in AlertDefs]
+        positiveRes = filter(lambda d: d[0], alertResults)
+        if positiveRes:
+            shortAlertMsgs = [x[1] for x in positiveRes]
+            _contentMsg = '\n\n'.join([
+                '*** THIS IS A GENERATED MESSAGE, PLEASE DO NOT REPLY ***',
+                'Workflow: {}'.format(doc['name']),
+                'Short Summary:\n{}'.format('\n'.join([
+                    '- {}'.format(s) for s in shortAlertMsgs
+                ])),
+                '-'* 79,
+                'Full document:\n{}'.format(
+                    json.dumps(
+                        doc, sort_keys=True, indent=4, separators=(',', ': ')))
+            ])
+
+            contentMsg = MIMEText(_contentMsg)
+            contentMsg['Subject'] = '[workflowmonit] Alert on * {} *'.format(
+                doc['name'])
+            contentMsg['From'] = sender
+            contentMsg['To'] = ', '.join(recipients)
+            s = smtplib.SMTP('localhost')
+            s.sendmail(sender, recipients, contentMsg.as_string())
+            s.quit()
+
+
+def errorEmailShooter(msg, recipients):
+    """
+    forward the error message to recipients by emails
+
+    :param msg str: error mesages
+    :param recipients list: list of recipients email address
+    """
+
+    sender = 'toolsandint-workflowmonitalert@cern.ch'
+
+    contentMsg = MIMEText(msg)
+    contentMsg['Subject'] = 'Exception caught for workflowmonit'
+    contentMsg['From'] = sender
+    contentMsg['To'] = ', '.join(recipients)
+    s = smtplib.SMTP('localhost')
+    s.sendmail(sender, recipients, contentMsg.as_string())
+    s.quit()
+
+
+def main():
+
+    import os
+    testdoc = os.path.join(
+        os.path.dirname(os.path.abspath(__file__)),
+        'Logs/toSendDoc_190317-033802.json')
+    docs = json.load(open(testdoc))
+    print([(d['name'], d['failureRate']) for d in docs])
+
+    alertWithEmail(docs, ['weinan.si@cern.ch', ])
+
+
+
+if __name__ == "__main__":
+    main()

workflowmonit/dumpWfStatusDb.py

@@ -1,4 +1,5 @@
 #!/user/bin/env python
+from __future__ import print_function
 import os
 import sys
 import sqlite3

workflowmonit/requirements.txt

@@ -1,3 +1,4 @@
 schedule
 git+https://github.com/dmwm/WMCore.git
 stomp.py
+pyyaml>=5.1

workflowmonit/sendToMonit.py

@@ -13,6 +13,7 @@
 import yaml
 from workflowmonit.stompAMQ import stompAMQ
 import workflowmonit.workflowCollector as wc
+import workflowmonit.alertingDefs as ad
 
 CRED_FILE_PATH = os.path.join(os.path.dirname(
     os.path.abspath(__file__)), 'credential.yml')
@@ -268,31 +269,40 @@ def sendDoc(cred, docs):
 
 def main():
 
-    with open(LOGGING_CONFIG, 'r') as f:
-        config = yaml.safe_load(f.read())
-        logging.config.dictConfig(config)
+    recipients = wc.get_yamlconfig(CONFIG_FILE_PATH).get('alert_recipients', [])
 
-    global logger
-    logger = logging.getLogger('workflowmonitLogger')
+    try:
+        with open(LOGGING_CONFIG, 'r') as f:
+            config = yaml.safe_load(f.read())
+            logging.config.dictConfig(config)
+
+        global logger
+        logger = logging.getLogger('workflowmonitLogger')
 
-    cred = wc.get_yamlconfig(CRED_FILE_PATH)
-    docs = buildDoc(CONFIG_FILE_PATH)
+        cred = wc.get_yamlconfig(CRED_FILE_PATH)
+        docs = buildDoc(CONFIG_FILE_PATH)
 
-    if not os.path.isdir(LOGDIR):
-        os.makedirs(LOGDIR)
+        # handling alerts
+        ad.alertWithEmail(docs, recipients)
 
-    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))
+        # backup documents
+        if not os.path.isdir(LOGDIR):
+            os.makedirs(LOGDIR)
 
-    failures = sendDoc(cred=cred, docs=docs)
+        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))
 
-    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))
+        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))
+    except Exception as e:
+        ad.errorEmailShooter(str(e), recipients)
 
 
 if __name__ == "__main__":

workflowmonit/stompAMQ.py

@@ -1,6 +1,4 @@
 #!/usr/bin/env python
-from __future__ import print_function
-
 import json
 
 from WMCore.Services.StompAMQ.StompAMQ import StompAMQ

workflowmonit/workflowCollector.py

@@ -44,7 +44,7 @@ def get_yamlconfig(configPath):
         return {}
 
     try:
-        return yaml.load(open(configPath).read())
+        return yaml.load(open(configPath).read(), Loader=yaml.FullLoader)
     except:
         return {}
 
@@ -290,7 +290,7 @@ def error_logs(workflow):
     if not wf_stepinfo:
         return error_logs
 
-    for stepname, stepdata in wf_stepinfo.iteritems():
+    for stepname, stepdata in wf_stepinfo.items():
         _taskName = stepname.split('/')[-1]
         # Get the errors from both 'jobfailed' and 'submitfailed' details
         for error, sitedata in [
@@ -301,7 +301,7 @@ def error_logs(workflow):
                 continue
             _errorcode = int(error)
 
-            for _sitename, siteinfo in sitedata.iteritems():
+            for _sitename, siteinfo in sitedata.items():
                 _errorsamples = list()
 
                 for sample in siteinfo['samples']:
@@ -383,18 +383,18 @@ def error_summary(workflow):
     if not errorInfo:
         return error_summary
 
-    for fullTaskName, taskErrors in errorInfo.iteritems():
+    for fullTaskName, taskErrors in errorInfo.items():
         taskName = fullTaskName.split('/')[-1]
         if not taskName:
             continue
 
         errorList = list()
         noReportSite = list(taskErrors.get('NotReported', {}).keys())
-        for errorCode, siteCnt in taskErrors.iteritems():
+        for errorCode, siteCnt in taskErrors.items():
             if errorCode == 'NotReported':
                 continue
 
-            for siteName, counts in siteCnt.iteritems():
+            for siteName, counts in siteCnt.items():
                 errorList.append({
                     'errorCode': int(errorCode),
                     'siteName': siteName,
@@ -457,16 +457,16 @@ def populate_error_for_workflow(workflow):
     workflow_summary['transitions'] = requestTransition
 
     nfailure = 0
-    for agent, agentdata in agentJobInfo.iteritems():
+    for agent, agentdata in agentJobInfo.items():
         status = agentdata.get('status', {})
         tasks = agentdata.get('tasks', {})
         if not all([status, tasks]):
             continue
 
-        for ftype, num in status.get('failure', {}).iteritems():
+        for ftype, num in status.get('failure', {}).items():
             nfailure += num
 
-        for taskFullName, taskData in tasks.iteritems():
+        for taskFullName, taskData in tasks.items():
             taskName = taskFullName.split('/')[-1]
 
             inputTask = None
@@ -479,13 +479,13 @@ def populate_error_for_workflow(workflow):
             taskSiteError = dict()
 
             if taskStatus and taskStatus.get('failure', {}):
-                for site, siteData in taskData.get('sites', {}).iteritems():
+                for site, siteData in taskData.get('sites', {}).items():
                     errCnt = 0
                     errCnts = siteData.get('failure', {})
                     if not errCnts:
                         continue
 
-                    for ftype, cnt in errCnts.iteritems():
+                    for ftype, cnt in errCnts.items():
                         errCnt += cnt
 
                     taskSiteError[site] = errCnt
@@ -499,7 +499,7 @@ def populate_error_for_workflow(workflow):
                 if 'siteErrors' not in _task.keys():
                     _task["siteErrors"] = taskSiteError
                 else:
-                    for site, errors in taskSiteError.iteritems():
+                    for site, errors in taskSiteError.items():
                         if site in _task["siteErrors"].keys():
                             _task["siteErrors"][site] += errors
                         else:
@@ -517,7 +517,7 @@ def populate_error_for_workflow(workflow):
 
     # remove tasks that does not have any error
     taskToDel = list()
-    for taskname, taskinfo in workflow_summary['tasks'].iteritems():
+    for taskname, taskinfo in workflow_summary['tasks'].items():
         if 'siteErrors' in taskinfo and (not taskinfo['siteErrors']):
             taskToDel.append(taskname)
     for taskname in taskToDel:
@@ -531,17 +531,17 @@ def populate_error_for_workflow(workflow):
         wf_errorLog = error_logs(workflow)
 
         # add information from errorSummary
-        for taskName, taskErrors in wf_errorSummary.iteritems():
+        for taskName, taskErrors in wf_errorSummary.items():
             if taskName in workflow_summary['tasks'].keys():
                 workflow_summary['tasks'][taskName].update(taskErrors)
 
         # add information from errorLog
-        for taskName, taskErrorLogInfo in wf_errorLog.iteritems():
+        for taskName, taskErrorLogInfo in wf_errorLog.items():
 
             if taskName not in workflow_summary['tasks'].keys():
                 continue
-            for errorCode, siteInfo in taskErrorLogInfo.iteritems():
-                for site, info in siteInfo.iteritems():
+            for errorCode, siteInfo in taskErrorLogInfo.items():
+                for site, info in siteInfo.items():
 
                     for e in workflow_summary['tasks'][taskName].get('errors', []):
                         if e.get('siteName', None) != site:
@@ -562,13 +562,13 @@ def populate_error_for_workflow(workflow):
 
     # last step, nest in task key(TaskName) as a key-value pair
     tasksAsList = []
-    for taskname, taskinfo in workflow_summary['tasks'].iteritems():
+    for taskname, taskinfo in workflow_summary['tasks'].items():
         taskinfo.update({"name": taskname})
         taskinfo['siteErrors'] = [
             {
                 "site": site,
                 "counts": counts
-            } for site, counts in taskinfo['siteErrors'].iteritems()
+            } for site, counts in taskinfo['siteErrors'].items()
         ]  # convert 'siteErrors' from a dict to a list of dict
         tasksAsList.append(taskinfo)
     workflow_summary['tasks'] = tasksAsList

workflowmonit/workflowMonitScheduler.py

@@ -1,6 +1,4 @@
 #!/usr/bin/env python
-from __future__ import print_function
-
 import os
 import time