pandajedi/jedidog/AtlasWorkflowManagerWatchDog.py

0001 import sys
0002 import traceback
0003
0004 # logger
0005 from pandacommon.pandalogger.PandaLogger import PandaLogger
0006
0007 from pandajedi.jedicore.MsgWrapper import MsgWrapper
0008 from pandaserver.workflow.workflow_core import WorkflowInterface
0009
0010 from .WatchDogBase import WatchDogBase
0011
0012 logger = PandaLogger().getLogger(__name__.split(".")[-1])
0013
0014
0015 class AtlasWorkflowManagerWatchDog(WatchDogBase):
0016     """
0017     Workflow manager watchdog for ATLAS
0018     """
0019
0020     # constructor
0021     def __init__(self, taskBufferIF, ddmIF):
0022         WatchDogBase.__init__(self, taskBufferIF, ddmIF)
0023         self.vo = "atlas"
0024         self.workflow_interface = WorkflowInterface(taskBufferIF)
0025
0026     def doProcessWorkflows(self):
0027         """
0028         Action to process active workflows
0029         """
0030         tmpLog = MsgWrapper(logger, " #ATM #KV doProcessWorkflows")
0031         tmpLog.debug("start")
0032         try:
0033             # watchdog lock
0034             got_lock = self.get_process_lock("AtlasWFManagerDog.doProcessWorkflows", timeLimit=5)
0035             if not got_lock:
0036                 tmpLog.debug("locked by another watchdog process. Skipped")
0037                 return
0038             tmpLog.debug("got watchdog lock")
0039             # process active workflows
0040             stats = self.workflow_interface.process_active_workflows()
0041             tmpLog.info(f"processed workflows: {stats}")
0042             # done
0043             tmpLog.debug("done")
0044         except Exception:
0045             errtype, errvalue = sys.exc_info()[:2]
0046             tmpLog.error(f"failed with {errtype} {errvalue} {traceback.format_exc()}")
0047
0048     # main
0049     def doAction(self):
0050         try:
0051             # get logger
0052             origTmpLog = MsgWrapper(logger)
0053             origTmpLog.debug("start")
0054             # clean up
0055             # check
0056             # process workflows
0057             self.doProcessWorkflows()
0058         except Exception:
0059             errtype, errvalue = sys.exc_info()[:2]
0060             origTmpLog.error(f"failed with {errtype} {errvalue}")
0061         # return
0062         origTmpLog.debug("done")
0063         return self.SC_SUCCEEDED